five-groups
/
five-groups-couplet
4
0
Fork 0
Code Issues Pull Requests 2 Packages Projects Releases Wiki Activity

修改bug

server_five_liuyunhu^2
lijiayao 2024-04-09 15:08:32 +08:00
parent 38521412a6
commit af273c12cb
4 changed files with 8 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java
View File

@ -38,7 +38,7 @@ import static java.lang.Thread.sleep;
@Component
@Slf4j
public class ModelsKafkaMessage {
private static final String TOPIC_NAME = "online";
private static final String TOPIC_NAME = "ljy";
private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";

10
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/RealTimeDataServiceImpl.java
View File

@ -41,17 +41,17 @@ public class RealTimeDataServiceImpl implements IncidentService {
public void incident(CoupletMsgData coupletMsgData) {
log.info("实时数据事件开始.....");
RealTimeDataRequest cacheObject = redisService.getCacheObject("vin:" + coupletMsgData.getVin());
//判断是否有缓存数据
if (redisService.hasKey("vin:query:" + coupletMsgData.getVin())){
redisService.deleteObject("vin:query:" + coupletMsgData.getVin());
}
// //判断是否有缓存数据
// if (redisService.hasKey("vin:query:" + coupletMsgData.getVin())){
// redisService.deleteObject("vin:query:" + coupletMsgData.getVin());
// }
// if (RealTimeJudge.isJudge(coupletMsgData.getVin())) {
if (coupletMsgData.getVin().equals(cacheObject.getVin())){
// log.info("有实时数据，值为：[{}]开始传输实时数据", coupletMsgData.getVin());
//判断数据是否一致，
// if (RealTimeJudge.addRealTime(cacheObject)) {
log.info("[{}]有缓存数据，值为：[{}]，且缓存数据与实时数据一致，开始传输实时数据", coupletMsgData.getVin(), cacheObject);
redisService.setCacheObject("vin:query:" + coupletMsgData.getVin(), cacheObject);
redisService.setCacheSet("vin:query:" + coupletMsgData.getVin(), coupletMsgData);
redisService.expire("vin:"+coupletMsgData.getVin(),10, TimeUnit.MINUTES);
// } else {
// log.info("[{}]有缓存数据，值为：[{}]，且缓存数据与实时数据不一致，开始传输实时数据", coupletMsgData.getVin(), cacheObject);

2
couplet-modules/couplet-modules-onLine/src/main/java/com/couplet/online/utils/MqttMonitor.java
View File

@ -76,7 +76,7 @@ public class MqttMonitor {
//Kafka生产者配置
private static final String TOPIC_NAME = "dong";
private static final String TOPIC_NAME = "ljy";
private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
//线程池，用于异步处理消息到来时的业务逻辑

2
couplet-modules/couplet-modules-onLine/src/main/resources/bootstrap.yml
View File

@ -36,7 +36,7 @@ mqtt:
# broker: mqtt://115.159.47.13:1883
username:
password:
clientId: xiaoYao
clientId: aaaaaaa
qos: 0
topic: xiaoYao