Skip to content

E
event
提交 690b55f0 作者: 925993793@qq.com
操作

热度计算

上级 491cb7ce
正在显示 包含 65 行增加 和 52 行删除
src/main/java/com/zzsn/event/constant/Constants.java
......@@ -8,14 +8,6 @@ public class Constants {
public static final String SUBJECT_ANALYSIS_PRE = "SUBJECT_ANALYSIS::";
//传播路径
public static final String PROPAGATION_KEY = "PROPAGATION_PATH::";
/**
* 关键词数据入缓存 key前缀
*/
public static final String KEY_WORDS_TO_REDIS_PREFIX = "KEY_WORDS_TO_REDIS::";
public static final String MAX_TOTAL_HOT_SCORE = "MAX_TOTAL_HOT_SCORE";
public static final String MAX_MEDIA_HOT_SCORE = "MAX_MEDIA_HOT_SCORE";
public static final String MAX_WECHAT_HOT_SCORE = "MAX_WECHAT_HOT_SCORE";
public static final String MAX_OTHER_HOT_SCORE = "MAX_OTHER_HOT_SCORE";
public static String[] FETCH_FIELDS_STATISTIC = {"id", "subjectId","title","origin", "publishDate", "sourceAddress"};
public static String[] FETCH_FIELDS_DATA = {"id", "subjectId", "title", "content", "publishDate", "origin", "sourceAddress"};
......
src/main/java/com/zzsn/event/kafka/KafkaConsumer.java
......@@ -42,7 +42,7 @@ public class KafkaConsumer {
*
* @param record 接受的kafka数据
*/
// @KafkaListener(topics = {Constants.VIEWPOINT_RECEIVE_TOPIC})
@KafkaListener(topics = {Constants.VIEWPOINT_RECEIVE_TOPIC})
public void viewPointAnalysis(ConsumerRecord<String, String> record) {
String value = record.value();
if (StringUtils.isNotEmpty(value)) {
......@@ -74,7 +74,7 @@ public class KafkaConsumer {
*
* @param record 接受的kafka数据
*/
// @KafkaListener(topics = {Constants.EVENT_CONTEXT_RECEIVE_TOPIC})
@KafkaListener(topics = {Constants.EVENT_CONTEXT_RECEIVE_TOPIC})
public void eventContext(ConsumerRecord<String, String> record) {
String value = record.value();
if (StringUtils.isNotEmpty(value)) {
......@@ -98,7 +98,7 @@ public class KafkaConsumer {
*
* @param record 接受的kafka数据
*/
// @KafkaListener(topics = {Constants.FAKE_EVENT_CONTEXT_RECEIVE_TOPIC})
@KafkaListener(topics = {Constants.FAKE_EVENT_CONTEXT_RECEIVE_TOPIC})
public void eventContext_fake(ConsumerRecord<String, String> record) {
String value = record.value();
if (StringUtils.isNotEmpty(value)) {
......@@ -132,7 +132,7 @@ public class KafkaConsumer {
* @author lkg
* @date 2024/4/12
*/
// @KafkaListener(topics = {Constants.EVENT_REPORT_RECEIVE_TOPIC})
@KafkaListener(topics = {Constants.EVENT_REPORT_RECEIVE_TOPIC})
public void eventReport(ConsumerRecord<String, String> record) {
String value = record.value();
EventAnalysisReport eventAnalysisReport = JSONObject.parseObject(value, EventAnalysisReport.class);
......
src/main/java/com/zzsn/event/task/EventHotTask.java
......@@ -3,21 +3,22 @@ package com.zzsn.event.task;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.zzsn.event.constant.Constants;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.zzsn.event.entity.Event;
import com.zzsn.event.service.EsStatisticsService;
import com.zzsn.event.service.IEventService;
import com.zzsn.event.util.RedisUtil;
import com.zzsn.event.util.CalculateUtil;
import com.zzsn.event.vo.HotVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
/**
*
......@@ -34,9 +35,6 @@ public class EventHotTask {
@Autowired
private EsStatisticsService esStatisticsService;
@Autowired
private RedisUtil redisUtil;
@Value("${scoreRule.weekScore}")
Integer weekScore;
@Value("${scoreRule.monthScore}")
......@@ -56,46 +54,53 @@ public class EventHotTask {
@Scheduled(cron = "0 0 0/3 * * ?")
public void hot() {
List<Event> list = eventService.list();
List<HotVO> countList = new ArrayList<>();
for (Event event : list) {
CompletableFuture.runAsync(() -> {
CompletableFuture<HotVO> async = CompletableFuture.supplyAsync(() -> {
HotVO hotVO = new HotVO();
String id = event.getId();
//total
Integer totalScore = computeScore(id);
Integer totalHot = getHot(Constants.MAX_TOTAL_HOT_SCORE + "::" + id, totalScore);
event.setTotalHot(totalHot);
/*//media
Integer mediaScore = 0;
Integer mediaHot = getHot(Constants.MAX_MEDIA_HOT_SCORE + "::" + id, mediaScore);
//wechat
Integer wechatScore = 0;
Integer wechatHot = getHot(Constants.MAX_WECHAT_HOT_SCORE + "::" + id, wechatScore);
//others
Integer otherScore = 0;
Integer otherHot = getHot(Constants.MAX_OTHER_HOT_SCORE + "::" + id, otherScore);
event.setMediaHot(mediaHot);
event.setWechatHot(wechatHot);
event.setOtherHot(otherHot);*/
int count = computeScore(id);
hotVO.setId(id);
hotVO.setCount(count);
return hotVO;
});
try {
HotVO hotVO = async.get();
countList.add(hotVO);
} catch (Exception e) {
e.printStackTrace();
}
}
countList.sort(Comparator.comparingInt(HotVO::getCount));
Integer min = countList.get(0).getCount();
Integer max = countList.get(countList.size() - 1).getCount();
Map<String, List<HotVO>> hotMap = countList.stream().collect(Collectors.groupingBy(HotVO::getId));
//最小-最大归一化(Min-Max Normalization)
for (Event event : list) {
List<HotVO> hotVOS = hotMap.get(event.getId());
if (CollectionUtils.isNotEmpty(hotVOS)) {
HotVO hotVO = hotVOS.get(0);
Integer hot = normalize(hotVO.getCount(), min, max);
event.setTotalHot(hot);
eventService.updateById(event);
log.info("专题-{}-热度计算完成", event.getEventName());
});
}
}
}
private Integer getHot(String key, Integer score) {
Object total = redisUtil.get(key);
if (null != total) {
Integer maxTotalScore = (Integer) total;
if (score > maxTotalScore) {
redisUtil.set(key, score);
return 100;
} else {
return score / maxTotalScore * 100;
}
} else {
redisUtil.set(key, score);
return 100;
}
//归一化计算
private Integer normalize(Integer value, Integer min,Integer max) {
int normalizeValue = 0;
if (value > 0) {
int param1 = (value - min) * 100;
int param2 = (max - min);
String divide = CalculateUtil.divide(String.valueOf(param1), String.valueOf(param2), 0);
if (StringUtils.isNotEmpty(divide)) {
normalizeValue = Integer.parseInt(divide);
}
}
return normalizeValue;
}
/**
......
src/main/java/com/zzsn/event/vo/HotVO.java 0 → 100644
package com.zzsn.event.vo;
import lombok.Data;
/**
* recursion
*
* @author lkg
* @date 2024/5/9
*/
@Data
public class HotVO {
private String id;
private Integer count;
}
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论