Middleware/cn-data-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Revert "冀北版本过滤无效数据"

Browse Source 
This reverts commit c1a2f3e8fb.
This commit is contained in:
hzj
2026-07-22 16:04:57 +08:00
parent c1a2f3e8fb
commit 14b213fb99
23 changed files with 1004 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
algorithm/algorithm-api/src/main/java/com/njcn/algorithm/pojo/api/PqReasonableRangeFeignClient.java
View File

@@ -12,7 +12,6 @@ import com.njcn.algorithm.pojo.api.fallback.PqReasonableRangeFeignClientFallback
import com.njcn.algorithm.pojo.dto.PqReasonableRangeDto; import com.njcn.algorithm.pojo.dto.PqReasonableRangeDto;
import com.njcn.algorithm.pojo.param.DataCleanParam; import com.njcn.algorithm.pojo.param.DataCleanParam;
import com.njcn.common.pojo.constant.ServerInfo; import com.njcn.common.pojo.constant.ServerInfo;
import com.njcn.common.pojo.response.HttpResult; import com.njcn.common.pojo.response.HttpResult;
import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiOperation;
import org.springframework.cloud.openfeign.FeignClient; import org.springframework.cloud.openfeign.FeignClient;

2
analysis/stat/stat-api/src/main/java/com/njcn/stat/messagedto/MessageHarmonicDataSet.java
View File

@@ -20,7 +20,7 @@ import java.time.LocalDateTime;
@Data @Data
public class MessageHarmonicDataSet implements Serializable { public class MessageHarmonicDataSet implements Serializable {
private Integer FLAG; private Integer FLAG;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss.SSS") @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
@JsonDeserialize(using = LocalDateTimeDeserializer.class) @JsonDeserialize(using = LocalDateTimeDeserializer.class)
@JsonSerialize(using = LocalDateTimeSerializer.class) @JsonSerialize(using = LocalDateTimeSerializer.class)
private LocalDateTime TIME; private LocalDateTime TIME;

3
analysis/stat/stat-boot/src/main/java/com/njcn/stat/service/impl/MessageAnalysisServiceImpl.java
View File

@@ -25,7 +25,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.temporal.ChronoField;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Objects; import java.util.Objects;
@@ -94,7 +93,7 @@ public class MessageAnalysisServiceImpl implements MessageAnalysisService {
MessageHarmonicDataSet messageHarmonicDataSet = JSONObject.parseObject(value, MessageHarmonicDataSet.class); MessageHarmonicDataSet messageHarmonicDataSet = JSONObject.parseObject(value, MessageHarmonicDataSet.class);
LocalDateTime localDateTime = messageHarmonicDataSet.getTIME(); LocalDateTime localDateTime = messageHarmonicDataSet.getTIME();
//排除上电下电等情况前置上送上不是整分的数据 //排除上电下电等情况前置上送上不是整分的数据
if((localDateTime.getSecond() != 0)||localDateTime.get(ChronoField.MILLI_OF_SECOND)!=0){ if(!(localDateTime.getSecond() == 0)){
return; return;
} }
Integer flag = messageHarmonicDataSet.getFLAG(); Integer flag = messageHarmonicDataSet.getFLAG();

6
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/LnDataDealController.java
View File

@@ -28,7 +28,6 @@ import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.temporal.ChronoField;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Objects; import java.util.Objects;
@@ -87,10 +86,7 @@ public class LnDataDealController extends BaseController {
if(Objects.equals(DataTypeEnum.HARMONIC.getCode(),dataType)) { if(Objects.equals(DataTypeEnum.HARMONIC.getCode(),dataType)) {
MessageHarmonicDataSet messageHarmonicDataSet = JSONObject.parseObject(value, MessageHarmonicDataSet.class); MessageHarmonicDataSet messageHarmonicDataSet = JSONObject.parseObject(value, MessageHarmonicDataSet.class);
LocalDateTime localDateTime = messageHarmonicDataSet.getTIME(); LocalDateTime localDateTime = messageHarmonicDataSet.getTIME();
//排除上电下电等情况前置上送上不是整分的数据
if((localDateTime.getSecond() != 0)||localDateTime.get(ChronoField.MILLI_OF_SECOND)!=0){
System.out.println(111111111);
}
Integer flag = messageHarmonicDataSet.getFLAG(); Integer flag = messageHarmonicDataSet.getFLAG();
MessageP pq = messageHarmonicDataSet.getPQ(); MessageP pq = messageHarmonicDataSet.getPQ();
MessageV v = messageHarmonicDataSet.getV(); MessageV v = messageHarmonicDataSet.getV();

12
message/message-boot/pom.xml
View File

@@ -105,12 +105,12 @@
<artifactId>mq-spring-boot-starter</artifactId> <artifactId>mq-spring-boot-starter</artifactId>
<version>1.0.0-SNAPSHOT</version> <version>1.0.0-SNAPSHOT</version>
</dependency> </dependency>
<!-- <dependency>--> <dependency>
<!-- <groupId>org.springframework.boot</groupId>--> <groupId>org.springframework.boot</groupId>
<!-- <artifactId>spring-boot-starter-test</artifactId>--> <artifactId>spring-boot-starter-test</artifactId>
<!-- <version>2.7.12</version>--> <version>2.7.12</version>
<!-- <scope>test</scope>--> <scope>test</scope>
<!-- </dependency>--> </dependency>
<dependency> <dependency>
<groupId>junit</groupId> <groupId>junit</groupId>
<artifactId>junit</artifactId> <artifactId>junit</artifactId>

97
message/message-boot/src/main/java/com/njcn/message/consumer/DeviceRunFlagDataConsumer.java
View File

@@ -6,7 +6,6 @@ import com.njcn.message.messagedto.DevComFlagDTO;
import com.njcn.message.constant.RedisKeyPrefix; import com.njcn.message.constant.RedisKeyPrefix;
import com.njcn.middle.rocket.constant.EnhanceMessageConstant; import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler; import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
import com.njcn.mq.annotation.MqListener;
import com.njcn.redis.pojo.enums.AppRedisKey; import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.pojo.enums.RedisKeyEnum; import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil; import com.njcn.redis.utils.RedisUtil;
@@ -31,18 +30,59 @@ import java.util.Objects;
* @version V1.0.0 * @version V1.0.0
*/ */
@Component @Component
@RocketMQMessageListener(
topic = "Device_Run_Flag_Topic",
consumerGroup = "Device_Run_Flag_Consumer",
selectorExpression = "*",
consumeThreadNumber = 10,
enableMsgTrace = true
)
@Slf4j @Slf4j
public class DeviceRunFlagDataConsumer { public class DeviceRunFlagDataConsumer extends EnhanceConsumerMessageHandler<DevComFlagDTO> implements RocketMQListener<String> {
@Resource @Resource
private RedisUtil redisUtil; private RedisUtil redisUtil;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Autowired @Autowired
private MessAnalysisFeignClient messAnalysisFeignClient; private MessAnalysisFeignClient messAnalysisFeignClient;
@Override
public void onMessage(String message) {
DevComFlagDTO devComFlagDTO = JSONObject.parseObject(message,DevComFlagDTO.class);
super.dispatchMessage(devComFlagDTO);
@MqListener(topic = "Device_Run_Flag_Topic", group = "Device_Run_Flag_Consumer") }
protected void onMessage(DevComFlagDTO message) {
/***
* 通过redis分布式锁判断当前消息所处状态
* 1、null 查不到该key的数据，属于第一次消费，放行
* 2、fail 上次消息消费时发生异常，放行
* 3、being processed 正在处理，打回去
* 4、success 最近72小时消费成功，避免重复消费，打回去
*/
@Override
public boolean filter(DevComFlagDTO message) {
// String keyStatus = redisUtil.getStringByKey(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()));
// if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.DEVICE_RUN_FLAG.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 30L);
// return false;
// }
return false;
}
/**
* 消费成功，缓存到redis72小时，避免重复消费
*/
@Override
protected void consumeSuccess(DevComFlagDTO message) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.DEVICE_RUN_FLAG.concat(message.getKey()), MessageStatus.SUCCESS, 5*60L);
}
@Override
protected void handleMessage(DevComFlagDTO message) {
//获取之前设备状态 //获取之前设备状态
//删除设备时前置会在连接一次通道但是DevId为空所以添加 //删除设备时前置会在连接一次通道但是DevId为空所以添加
if(StringUtils.isNoneBlank(message.getId())){ if(StringUtils.isNoneBlank(message.getId())){
@@ -62,6 +102,53 @@ public class DeviceRunFlagDataConsumer {
/**
* 发生异常时，进行错误信息入库保存
* 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
*/
@Override
protected void saveExceptionMsgLog(DevComFlagDTO message, String identity, Exception exception) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.DEVICE_RUN_FLAG.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
rocketmqMsgErrorLog.setMsgKey(message.getKey());
rocketmqMsgErrorLog.setResource(message.getSource());
if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
//数据库字段配置长度200，避免插入失败，大致分析异常原因
String exceptionMsg = exception.getMessage();
if(exceptionMsg.length() > 200){
exceptionMsg = exceptionMsg.substring(0,180);
}
rocketmqMsgErrorLog.setRecord(exceptionMsg);
//如果是当前消息重试的则略过
if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
//单次消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
} else {
rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
//重试N次后，依然消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
}
/***
* 处理失败后，是否重试
* 一般开启
*/
@Override
protected boolean isRetry() {
return true;
}
/***
* 消费失败是否抛出异常，抛出异常后就不再消费了
*/
@Override
protected boolean throwException() {
return false;
}

95
message/message-boot/src/main/java/com/njcn/message/consumer/FileSysDataConsumer.java
View File

@@ -8,7 +8,6 @@ import com.njcn.message.messagedto.MessageDataDTO;
import com.njcn.message.websocket.WebSocketServer; import com.njcn.message.websocket.WebSocketServer;
import com.njcn.middle.rocket.constant.EnhanceMessageConstant; import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler; import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
import com.njcn.mq.annotation.MqListener;
import com.njcn.redis.pojo.enums.RedisKeyEnum; import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil; import com.njcn.redis.utils.RedisUtil;
import com.njcn.system.api.RocketMqLogFeignClient; import com.njcn.system.api.RocketMqLogFeignClient;
@@ -32,17 +31,59 @@ import java.util.concurrent.TimeUnit;
* @version V1.0.0 * @version V1.0.0
*/ */
@Component @Component
@RocketMQMessageListener(
topic = "File_Reply_Topic",
consumerGroup = "file_sys_consumer",
consumeThreadNumber = 10,
enableMsgTrace = true
)
@Slf4j @Slf4j
public class FileSysDataConsumer { public class FileSysDataConsumer extends EnhanceConsumerMessageHandler<FileSysDTO> implements RocketMQListener<String> {
@Resource @Resource
private RedisUtil redisUtil; private RedisUtil redisUtil;
@Resource @Resource
private StringRedisTemplate stringRedisTemplate; private StringRedisTemplate stringRedisTemplate;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Override
public void onMessage(String message) {
@MqListener(topic = "File_Reply_Topic", group = "file_sys_consumer") FileSysDTO messageDataDTO = JSONObject.parseObject(message, FileSysDTO.class);
super.dispatchMessage(messageDataDTO);
}
/***
* 通过redis分布式锁判断当前消息所处状态
* 1、null 查不到该key的数据，属于第一次消费，放行
* 2、fail 上次消息消费时发生异常，放行
* 3、being processed 正在处理，打回去
* 4、success 最近72小时消费成功，避免重复消费，打回去
*/
@Override
public boolean filter(FileSysDTO message) {
String keyStatus = redisUtil.getStringByKey(RedisKeyPrefix.REAL_TIME_DATA.concat(message.getKey()));
if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.REAL_TIME_DATA.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 30L);
return false;
}
return true;
}
/**
* 消费成功，缓存到redis5分钟，避免重复消费
*/
@Override
protected void consumeSuccess(FileSysDTO message) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.REAL_TIME_DATA.concat(message.getKey()), MessageStatus.SUCCESS, 5*60L);
}
@Override
protected void handleMessage(FileSysDTO message) { protected void handleMessage(FileSysDTO message) {
String msgId = message.getGuid(); String msgId = message.getGuid();
@@ -56,8 +97,54 @@ public class FileSysDataConsumer {
/**
* 发生异常时，进行错误信息入库保存
* 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
*/
@Override
protected void saveExceptionMsgLog(FileSysDTO message, String identity, Exception exception) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.REAL_TIME_DATA.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
rocketmqMsgErrorLog.setMsgKey(message.getKey());
rocketmqMsgErrorLog.setResource(message.getSource());
if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
//数据库字段配置长度200，避免插入失败，大致分析异常原因
String exceptionMsg = exception.getMessage();
if(exceptionMsg.length() > 200){
exceptionMsg = exceptionMsg.substring(0,180);
}
rocketmqMsgErrorLog.setRecord(exceptionMsg);
//如果是当前消息重试的则略过
if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
//单次消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
} else {
rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
//重试N次后，依然消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
}
/***
* 处理失败后，是否重试
* 一般开启
*/
@Override
protected boolean isRetry() {
return true;
}
/***
* 消费失败是否抛出异常，抛出异常后就不再消费了
*/
@Override
protected boolean throwException() {
return false;
}

167
message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumer.java
View File

@@ -1,18 +1,34 @@
package com.njcn.message.consumer; package com.njcn.message.consumer;
import com.alibaba.fastjson.JSONObject;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.message.constant.MessageStatus;
import com.njcn.message.messagedto.MessageDataDTO; import com.njcn.message.messagedto.MessageDataDTO;
import com.njcn.mq.annotation.MqListener; import com.njcn.message.constant.RedisKeyPrefix;
import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.stat.api.MessAnalysisFeignClient; import com.njcn.stat.api.MessAnalysisFeignClient;
import com.njcn.system.api.RocketMqLogFeignClient;
import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/** /**
@@ -23,20 +39,151 @@ import java.util.List;
* @createTime 2023/8/11 15:32 * @createTime 2023/8/11 15:32
*/ */
@Component @Component
@org.springframework.boot.autoconfigure.condition.ConditionalOnProperty(
name = "mq.type", havingValue = "rocketmq", matchIfMissing = true)
@RocketMQMessageListener(
topic = "LN_Topic",
consumerGroup = "ln_consumer",
selectorExpression = "*",
consumeThreadNumber = 10,
enableMsgTrace = true
)
@Slf4j @Slf4j
public class FrontDataConsumer { public class FrontDataConsumer extends EnhanceConsumerMessageHandler<MessageDataDTO> implements RocketMQListener<String> {
@Autowired @Autowired
private MessAnalysisFeignClient messAnalysisFeignClient; private MessAnalysisFeignClient messAnalysisFeignClient;
@MqListener(topic = "MQ_INST_1697676490574298_Bm3DLamM%LN_Topic", group = "ln_consumer",batchSize = 50,batchTimeoutMs=10000) @Value("${rocketmq.consumer_size}")
public void onMessage(List<MessageDataDTO> msg) { private Integer consumerSize;
log.info("开始消费");
messAnalysisFeignClient.analysis(msg); @Resource
private RedisUtil redisUtil;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
private List<MessageDataDTO> messageList = new ArrayList<>();
@PostConstruct
public void validateConfig() {
if (consumerSize == null) {
throw new IllegalStateException("rocketmq.consumer_size 未配置！");
}
this.messageList = new ArrayList<>(consumerSize);
}
@Override
public void onMessage(String baseMessage) {
MessageDataDTO messageDataDTO = JSONObject.parseObject(baseMessage,MessageDataDTO.class);
super.dispatchMessage(messageDataDTO);
}
/***
* 通过redis分布式锁判断当前消息所处状态
* 1、null 查不到该key的数据，属于第一次消费，放行
* 2、fail 上次消息消费时发生异常，放行
* 3、being processed 正在处理，打回去
* 4、success 最近72小时消费成功，避免重复消费，打回去
*/
@Override
public boolean filter(MessageDataDTO message) {
String keyStatus = redisUtil.getStringByKey(message.getKey());
if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.HARMMONIC_TOPIC.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 60L);
return false;
}
return true;
}
/**
* 消费成功，缓存到redis5分钟，避免重复消费
*/
@Override
protected void consumeSuccess(MessageDataDTO message) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.HARMMONIC_TOPIC.concat(message.getKey()), MessageStatus.SUCCESS, 5*60L);
}
@Override
protected void handleMessage(MessageDataDTO message) {
synchronized (messageList) {
messageList.add(message);
if (messageList.size() >= consumerSize) {
saveToDatabase();
}
}
} }
/**
* 发生异常时，进行错误信息入库保存
* 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
*/
@Override
protected void saveExceptionMsgLog(MessageDataDTO message, String identity, Exception exception) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.HARMMONIC_TOPIC.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
rocketmqMsgErrorLog.setMsgKey(message.getKey());
rocketmqMsgErrorLog.setResource(message.getSource());
if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
//数据库字段配置长度200，避免插入失败，大致分析异常原因
String exceptionMsg = exception.getMessage();
if(exceptionMsg.length() > 200){
exceptionMsg = exceptionMsg.substring(0,180);
}
rocketmqMsgErrorLog.setRecord(exceptionMsg);
//如果是当前消息重试的则略过
if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
//单次消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
} else {
rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
//重试N次后，依然消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
}
/***
* 处理失败后，是否重试
* 一般开启
*/
@Override
protected boolean isRetry() {
return false;
}
/***
* 消费失败是否抛出异常，抛出异常后就不再消费了
*/
@Override
protected boolean throwException() {
return false;
}
//50个消息做一组插入数据库
public void saveToDatabase(){
try {
long start = System.currentTimeMillis();
messAnalysisFeignClient.analysis(messageList);
long end = System.currentTimeMillis();
log.info("处理"+consumerSize+"条消息所需时间------------"+(end-start));
}catch (Exception e){{
log.info(e.toString());
}
}finally{
messageList.clear();
}
}
} }

96
message/message-boot/src/main/java/com/njcn/message/consumer/FrontHeartBeatConsumer.java
View File

@@ -7,7 +7,6 @@ import com.njcn.message.messagedto.FrontHeartBeatDTO;
import com.njcn.message.constant.RedisKeyPrefix; import com.njcn.message.constant.RedisKeyPrefix;
import com.njcn.middle.rocket.constant.EnhanceMessageConstant; import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler; import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
import com.njcn.mq.annotation.MqListener;
import com.njcn.redis.pojo.enums.AppRedisKey; import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.pojo.enums.RedisKeyEnum; import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil; import com.njcn.redis.utils.RedisUtil;
@@ -31,17 +30,58 @@ import java.util.Objects;
* @version V1.0.0 * @version V1.0.0
*/ */
@Component @Component
@RocketMQMessageListener(
topic = "Heart_Beat_Topic",
consumerGroup = "Heartb_Beat_Topic_Consumer",
selectorExpression = "*",
consumeThreadNumber = 10,
enableMsgTrace = true
)
@Slf4j @Slf4j
public class FrontHeartBeatConsumer { public class FrontHeartBeatConsumer extends EnhanceConsumerMessageHandler<FrontHeartBeatDTO> implements RocketMQListener<String> {
@Resource @Resource
private RedisUtil redisUtil; private RedisUtil redisUtil;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Autowired @Autowired
private MessAnalysisFeignClient messAnalysisFeignClient; private MessAnalysisFeignClient messAnalysisFeignClient;
@MqListener(topic = "Heart_Beat_Topic", group = "Heartb_Beat_Topic_Consumer") @Override
public void onMessage(String message) {
FrontHeartBeatDTO frontHeartBeatDTO = JSONObject.parseObject(message, FrontHeartBeatDTO.class);
super.dispatchMessage(frontHeartBeatDTO);
}
//本消息不需要控制重复消费
/***
* 通过redis分布式锁判断当前消息所处状态
* 1、null 查不到该key的数据，属于第一次消费，放行
* 2、fail 上次消息消费时发生异常，放行
* 3、being processed 正在处理，打回去
* 4、success 最近72小时消费成功，避免重复消费，打回去
*/
@Override
public boolean filter(FrontHeartBeatDTO message) {
// String keyStatus = redisUtil.getStringByKey(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()));
// if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.HEART_BEAT.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 30L);
// return false;
// }
return false;
}
/**
* 消费成功，缓存到redis72小时，避免重复消费
*/
@Override
protected void consumeSuccess(FrontHeartBeatDTO message) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.HEART_BEAT.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
}
@Override
protected void handleMessage(FrontHeartBeatDTO message) { protected void handleMessage(FrontHeartBeatDTO message) {
//将心跳状态存到redis失效时间为30s，如果持续有心跳则进程在线反之，redis找不到则不在线 //将心跳状态存到redis失效时间为30s，如果持续有心跳则进程在线反之，redis找不到则不在线
if(Objects.equals(message.getFronttype(), FrontTypeEnum.STAT.getCode())){ if(Objects.equals(message.getFronttype(), FrontTypeEnum.STAT.getCode())){
@@ -53,8 +93,54 @@ public class FrontHeartBeatConsumer {
/**
* 发生异常时，进行错误信息入库保存
* 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
*/
@Override
protected void saveExceptionMsgLog(FrontHeartBeatDTO message, String identity, Exception exception) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.HEART_BEAT.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
rocketmqMsgErrorLog.setMsgKey(message.getKey());
rocketmqMsgErrorLog.setResource(message.getSource());
if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
//数据库字段配置长度200，避免插入失败，大致分析异常原因
String exceptionMsg = exception.getMessage();
if(exceptionMsg.length() > 200){
exceptionMsg = exceptionMsg.substring(0,180);
}
rocketmqMsgErrorLog.setRecord(exceptionMsg);
//如果是当前消息重试的则略过
if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
//单次消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
} else {
rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
//重试N次后，依然消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
}
/***
* 处理失败后，是否重试
* 一般开启
*/
@Override
protected boolean isRetry() {
return true;
}
/***
* 消费失败是否抛出异常，抛出异常后就不再消费了
*/
@Override
protected boolean throwException() {
return false;
}

94
message/message-boot/src/main/java/com/njcn/message/consumer/RealTimeDataConsumer.java
View File

@@ -8,7 +8,6 @@ import com.njcn.message.constant.RedisKeyPrefix;
import com.njcn.message.websocket.WebSocketServer; import com.njcn.message.websocket.WebSocketServer;
import com.njcn.middle.rocket.constant.EnhanceMessageConstant; import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler; import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
import com.njcn.mq.annotation.MqListener;
import com.njcn.redis.pojo.enums.RedisKeyEnum; import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil; import com.njcn.redis.utils.RedisUtil;
import com.njcn.system.api.RocketMqLogFeignClient; import com.njcn.system.api.RocketMqLogFeignClient;
@@ -29,16 +28,57 @@ import java.util.Objects;
* @version V1.0.0 * @version V1.0.0
*/ */
@Component @Component
@RocketMQMessageListener(
topic = "Real_Time_Data_Topic",
consumerGroup = "real_time_consumer",
selectorExpression = "*",
consumeThreadNumber = 10,
enableMsgTrace = true
)
@Slf4j @Slf4j
public class RealTimeDataConsumer { public class RealTimeDataConsumer extends EnhanceConsumerMessageHandler<MessageDataDTO> implements RocketMQListener<String> {
@Resource @Resource
private RedisUtil redisUtil; private RedisUtil redisUtil;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Override
public void onMessage(String message) {
MessageDataDTO messageDataDTO = JSONObject.parseObject(message,MessageDataDTO.class);
super.dispatchMessage(messageDataDTO);
}
@MqListener(topic = "Real_Time_Data_Topic", group = "real_time_consumer")
/***
* 通过redis分布式锁判断当前消息所处状态
* 1、null 查不到该key的数据，属于第一次消费，放行
* 2、fail 上次消息消费时发生异常，放行
* 3、being processed 正在处理，打回去
* 4、success 最近72小时消费成功，避免重复消费，打回去
*/
@Override
public boolean filter(MessageDataDTO message) {
String keyStatus = redisUtil.getStringByKey(RedisKeyPrefix.REAL_TIME_DATA.concat(message.getKey()));
if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.REAL_TIME_DATA.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 30L);
return false;
}
return true;
}
/**
* 消费成功，缓存到redis5分钟，避免重复消费
*/
@Override
protected void consumeSuccess(MessageDataDTO message) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.REAL_TIME_DATA.concat(message.getKey()), MessageStatus.SUCCESS, 5*60L);
}
@Override
protected void handleMessage(MessageDataDTO message) { protected void handleMessage(MessageDataDTO message) {
String lineId = message.getMonitor(); String lineId = message.getMonitor();
WebSocketServer.sendInfo(JSONObject.toJSONString(message),lineId); WebSocketServer.sendInfo(JSONObject.toJSONString(message),lineId);
@@ -49,8 +89,54 @@ public class RealTimeDataConsumer {
/**
* 发生异常时，进行错误信息入库保存
* 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
*/
@Override
protected void saveExceptionMsgLog(MessageDataDTO message, String identity, Exception exception) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.REAL_TIME_DATA.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
rocketmqMsgErrorLog.setMsgKey(message.getKey());
rocketmqMsgErrorLog.setResource(message.getSource());
if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
//数据库字段配置长度200，避免插入失败，大致分析异常原因
String exceptionMsg = exception.getMessage();
if(exceptionMsg.length() > 200){
exceptionMsg = exceptionMsg.substring(0,180);
}
rocketmqMsgErrorLog.setRecord(exceptionMsg);
//如果是当前消息重试的则略过
if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
//单次消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
} else {
rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
//重试N次后，依然消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
}
/***
* 处理失败后，是否重试
* 一般开启
*/
@Override
protected boolean isRetry() {
return true;
}
/***
* 消费失败是否抛出异常，抛出异常后就不再消费了
*/
@Override
protected boolean throwException() {
return false;
}

96
message/message-boot/src/main/java/com/njcn/message/consumer/TopicLogsConsumer.java
View File

@@ -6,7 +6,6 @@ import com.njcn.message.constant.RedisKeyPrefix;
import com.njcn.message.messagedto.FrontLogslMessage; import com.njcn.message.messagedto.FrontLogslMessage;
import com.njcn.middle.rocket.constant.EnhanceMessageConstant; import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler; import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
import com.njcn.mq.annotation.MqListener;
import com.njcn.redis.pojo.enums.AppRedisKey; import com.njcn.redis.pojo.enums.AppRedisKey;
import com.njcn.redis.pojo.enums.RedisKeyEnum; import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil; import com.njcn.redis.utils.RedisUtil;
@@ -31,15 +30,58 @@ import java.util.Objects;
* @version V1.0.0 * @version V1.0.0
*/ */
@Component @Component
@RocketMQMessageListener(
topic = "log_Topic",
consumerGroup = "Log_Topic_Consumer",
selectorExpression = "*",
consumeThreadNumber = 10,
enableMsgTrace = true
)
@Slf4j @Slf4j
public class TopicLogsConsumer { public class TopicLogsConsumer extends EnhanceConsumerMessageHandler<FrontLogslMessage> implements RocketMQListener<String> {
@Resource @Resource
private RedisUtil redisUtil; private RedisUtil redisUtil;
@Resource @Resource
private FrontLogsFeignClient frontLogsFeignClient; private FrontLogsFeignClient frontLogsFeignClient;
@MqListener(topic = "log_Topic", group = "Log_Topic_Consumer") @Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Override
public void onMessage(String message) {
FrontLogslMessage frontLogslMessage = JSONObject.parseObject(message,FrontLogslMessage.class);
super.dispatchMessage(frontLogslMessage);
}
/***
* 通过redis分布式锁判断当前消息所处状态
* 1、null 查不到该key的数据，属于第一次消费，放行
* 2、fail 上次消息消费时发生异常，放行
* 3、being processed 正在处理，打回去
* 4、success 最近72小时消费成功，避免重复消费，打回去
*/
@Override
public boolean filter(FrontLogslMessage message) {
// String keyStatus = redisUtil.getStringByKey(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()));
// if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.TOPIC_REPLY.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 30L);
// return false;
// }
return false;
}
/**
* 消费成功，缓存到redis72小时，避免重复消费
*/
@Override
protected void consumeSuccess(FrontLogslMessage message) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.TOPIC_REPLY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
}
@Override
protected void handleMessage(FrontLogslMessage message) { protected void handleMessage(FrontLogslMessage message) {
//业务处理 //业务处理
@@ -51,8 +93,54 @@ public class TopicLogsConsumer {
/**
* 发生异常时，进行错误信息入库保存
* 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
*/
@Override
protected void saveExceptionMsgLog(FrontLogslMessage message, String identity, Exception exception) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.TOPIC_REPLY.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
rocketmqMsgErrorLog.setMsgKey(message.getKey());
rocketmqMsgErrorLog.setResource(message.getSource());
if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
//数据库字段配置长度200，避免插入失败，大致分析异常原因
String exceptionMsg = exception.getMessage();
if(exceptionMsg.length() > 200){
exceptionMsg = exceptionMsg.substring(0,180);
}
rocketmqMsgErrorLog.setRecord(exceptionMsg);
//如果是当前消息重试的则略过
if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
//单次消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
} else {
rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
//重试N次后，依然消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
}
/***
* 处理失败后，是否重试
* 一般开启
*/
@Override
protected boolean isRetry() {
return false;
}
/***
* 消费失败是否抛出异常，抛出异常后就不再消费了
*/
@Override
protected boolean throwException() {
return false;
}

95
message/message-boot/src/main/java/com/njcn/message/consumer/TopicReplyConsumer.java
View File

@@ -7,7 +7,6 @@ import com.njcn.message.messagedto.TopicReplyDTO;
import com.njcn.message.constant.RedisKeyPrefix; import com.njcn.message.constant.RedisKeyPrefix;
import com.njcn.middle.rocket.constant.EnhanceMessageConstant; import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler; import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
import com.njcn.mq.annotation.MqListener;
import com.njcn.redis.pojo.enums.RedisKeyEnum; import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisUtil; import com.njcn.redis.utils.RedisUtil;
import com.njcn.stat.api.MessAnalysisFeignClient; import com.njcn.stat.api.MessAnalysisFeignClient;
@@ -38,14 +37,50 @@ import java.util.Objects;
enableMsgTrace = true enableMsgTrace = true
) )
@Slf4j @Slf4j
public class TopicReplyConsumer { public class TopicReplyConsumer extends EnhanceConsumerMessageHandler<TopicReplyDTO> implements RocketMQListener<String> {
@Resource @Resource
private RedisUtil redisUtil; private RedisUtil redisUtil;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Autowired @Autowired
private MessAnalysisFeignClient messAnalysisFeignClient; private MessAnalysisFeignClient messAnalysisFeignClient;
@MqListener(topic = "Topic_Reply_Topic", group = "Topic_Reply_Topic_Consumer") @Override
public void onMessage(String message) {
TopicReplyDTO topicReplyDTO = JSONObject.parseObject(message,TopicReplyDTO.class);
super.dispatchMessage(topicReplyDTO);
}
/***
* 通过redis分布式锁判断当前消息所处状态
* 1、null 查不到该key的数据，属于第一次消费，放行
* 2、fail 上次消息消费时发生异常，放行
* 3、being processed 正在处理，打回去
* 4、success 最近72小时消费成功，避免重复消费，打回去
*/
@Override
public boolean filter(TopicReplyDTO message) {
// String keyStatus = redisUtil.getStringByKey(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()));
// if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.TOPIC_REPLY.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 30L);
// return false;
// }
return false;
}
/**
* 消费成功，缓存到redis72小时，避免重复消费
*/
@Override
protected void consumeSuccess(TopicReplyDTO message) {
// redisUtil.saveByKeyWithExpire(RedisKeyPrefix.TOPIC_REPLY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
}
@Override
protected void handleMessage(TopicReplyDTO message) { protected void handleMessage(TopicReplyDTO message) {
//“12345”补招回复 //“12345”补招回复
if(Objects.equals(message.getGuid(),"12345")){ if(Objects.equals(message.getGuid(),"12345")){
@@ -59,4 +94,56 @@ public class TopicReplyConsumer {
/**
* 发生异常时，进行错误信息入库保存
* 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
*/
@Override
protected void saveExceptionMsgLog(TopicReplyDTO message, String identity, Exception exception) {
redisUtil.saveByKeyWithExpire(RedisKeyPrefix.TOPIC_REPLY.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
rocketmqMsgErrorLog.setMsgKey(message.getKey());
rocketmqMsgErrorLog.setResource(message.getSource());
if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
//数据库字段配置长度200，避免插入失败，大致分析异常原因
String exceptionMsg = exception.getMessage();
if(exceptionMsg.length() > 200){
exceptionMsg = exceptionMsg.substring(0,180);
}
rocketmqMsgErrorLog.setRecord(exceptionMsg);
//如果是当前消息重试的则略过
if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
//单次消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
} else {
rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
//重试N次后，依然消费异常
rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
}
}
/***
* 处理失败后，是否重试
* 一般开启
*/
@Override
protected boolean isRetry() {
return true;
}
/***
* 消费失败是否抛出异常，抛出异常后就不再消费了
*/
@Override
protected boolean throwException() {
return false;
}
} }

46
message/message-boot/src/main/java/com/njcn/message/mq/FrontDataMqListener.java
View File

@@ -1,23 +1,23 @@
//package com.njcn.message.mq; package com.njcn.message.mq;
//
//import com.njcn.message.messagedto.MessageDataDTO; import com.njcn.message.messagedto.MessageDataDTO;
//import com.njcn.mq.annotation.MqListener; import com.njcn.mq.annotation.MqListener;
//import com.njcn.stat.api.MessAnalysisFeignClient; import com.njcn.stat.api.MessAnalysisFeignClient;
//import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
//import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource; import javax.annotation.Resource;
//import java.util.Collections; import java.util.Collections;
//
//@Component @Component
//@ConditionalOnProperty(name = "mq.type", havingValue = "redis-stream") @ConditionalOnProperty(name = "mq.type", havingValue = "redis-stream")
//public class FrontDataMqListener { public class FrontDataMqListener {
//
// @Resource @Resource
// private MessAnalysisFeignClient messAnalysisFeignClient; private MessAnalysisFeignClient messAnalysisFeignClient;
//
// @MqListener(topic = "LN_Topic", group = "ln_consumer") @MqListener(topic = "LN_Topic", group = "ln_consumer")
// public void onFrontData(MessageDataDTO msg) { public void onFrontData(MessageDataDTO msg) {
// messAnalysisFeignClient.analysis(Collections.singletonList(msg)); messAnalysisFeignClient.analysis(Collections.singletonList(msg));
// } }
//} }

38
message/message-boot/src/main/java/com/njcn/message/mq/RecallMqSender.java
View File

@@ -1,19 +1,19 @@
//package com.njcn.message.mq; package com.njcn.message.mq;
//
//import com.njcn.message.constant.BusinessTopic; import com.njcn.message.constant.BusinessTopic;
//import com.njcn.message.message.RecallMessage; import com.njcn.message.message.RecallMessage;
//import com.njcn.mq.core.MqTemplate; import com.njcn.mq.core.MqTemplate;
//import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource; import javax.annotation.Resource;
//
//@Component @Component
//public class RecallMqSender { public class RecallMqSender {
//
// @Resource @Resource
// private MqTemplate mqTemplate; private MqTemplate mqTemplate;
//
// public void send(RecallMessage message, String nodeId) { public void send(RecallMessage message, String nodeId) {
// mqTemplate.send(nodeId + "_" + BusinessTopic.RECALL_TOPIC, message); mqTemplate.send(nodeId + "_" + BusinessTopic.RECALL_TOPIC, message);
// } }
//} }

2
message/message-boot/src/main/java/com/njcn/message/produce/ProduceController.java
View File

@@ -86,7 +86,7 @@ public class ProduceController extends BaseController {
@ApiImplicitParam(name = "message", value = "参数", required = true) @ApiImplicitParam(name = "message", value = "参数", required = true)
public HttpResult<String> askFileSys(@RequestBody AskFileSysMessage message){ public HttpResult<String> askFileSys(@RequestBody AskFileSysMessage message){
String methodDescribe = getMethodDescribe("askFileSys"); String methodDescribe = getMethodDescribe("askFileSys");
askFileSysMessaggeTemplate.sendMember(message); SendResult sendResult = askFileSysMessaggeTemplate.sendMember(message);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS,message.getGuid() , methodDescribe); return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS,message.getGuid() , methodDescribe);
} }
} }

15
message/message-boot/src/main/java/com/njcn/message/produce/template/AskFileSysMessaggeTemplate.java
View File

@@ -7,13 +7,10 @@ import com.njcn.message.constant.BusinessTopic;
import com.njcn.message.message.AskFileSysMessage; import com.njcn.message.message.AskFileSysMessage;
import com.njcn.middle.rocket.domain.BaseMessage; import com.njcn.middle.rocket.domain.BaseMessage;
import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate; import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate;
import com.njcn.mq.core.MqTemplate;
import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/** /**
* Description: * Description:
* Date: 2024/12/13 15:15【需求编号】 * Date: 2024/12/13 15:15【需求编号】
@@ -22,14 +19,16 @@ import javax.annotation.Resource;
* @version V1.0.0 * @version V1.0.0
*/ */
@Component @Component
public class AskFileSysMessaggeTemplate { public class AskFileSysMessaggeTemplate extends RocketMQEnhanceTemplate {
@Resource public AskFileSysMessaggeTemplate(RocketMQTemplate template) {
private MqTemplate mqTemplate; super(template);
public void sendMember(AskFileSysMessage askRealDataMessage) { }
public SendResult sendMember(AskFileSysMessage askRealDataMessage) {
BaseMessage baseMessage = new BaseMessage(); BaseMessage baseMessage = new BaseMessage();
baseMessage.setMessageBody(JSONObject.toJSONString(askRealDataMessage)); baseMessage.setMessageBody(JSONObject.toJSONString(askRealDataMessage));
baseMessage.setSource(BusinessResource.WEB_RESOURCE); baseMessage.setSource(BusinessResource.WEB_RESOURCE);
baseMessage.setKey(askRealDataMessage.getGuid()); baseMessage.setKey(askRealDataMessage.getGuid());
mqTemplate.send(BusinessTopic.FILE_TOPIC,askRealDataMessage.getNodeId() , baseMessage); return send(BusinessTopic.FILE_TOPIC,askRealDataMessage.getNodeId() , baseMessage);
} }
} }

15
message/message-boot/src/main/java/com/njcn/message/produce/template/AskRealDataMessaggeTemplate.java
View File

@@ -7,13 +7,10 @@ import com.njcn.message.constant.BusinessTopic;
import com.njcn.message.message.AskRealDataMessage; import com.njcn.message.message.AskRealDataMessage;
import com.njcn.middle.rocket.domain.BaseMessage; import com.njcn.middle.rocket.domain.BaseMessage;
import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate; import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate;
import com.njcn.mq.core.MqTemplate;
import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/** /**
* Description: * Description:
* Date: 2024/12/13 15:15【需求编号】 * Date: 2024/12/13 15:15【需求编号】
@@ -22,13 +19,15 @@ import javax.annotation.Resource;
* @version V1.0.0 * @version V1.0.0
*/ */
@Component @Component
public class AskRealDataMessaggeTemplate { public class AskRealDataMessaggeTemplate extends RocketMQEnhanceTemplate {
@Resource public AskRealDataMessaggeTemplate(RocketMQTemplate template) {
private MqTemplate mqTemplate; super(template);
public void sendMember(BaseMessage askRealDataMessage,String nodeId) { }
public SendResult sendMember(BaseMessage askRealDataMessage,String nodeId) {
askRealDataMessage.setSource(BusinessResource.WEB_RESOURCE); askRealDataMessage.setSource(BusinessResource.WEB_RESOURCE);
AskRealDataMessage dto = JSON.parseObject(askRealDataMessage.getMessageBody(), AskRealDataMessage.class); AskRealDataMessage dto = JSON.parseObject(askRealDataMessage.getMessageBody(), AskRealDataMessage.class);
askRealDataMessage.setKey(dto.getLine()); askRealDataMessage.setKey(dto.getLine());
mqTemplate.send(BusinessTopic.ASK_REAL_DATA_TOPIC,nodeId , askRealDataMessage); return send(BusinessTopic.ASK_REAL_DATA_TOPIC,nodeId , askRealDataMessage);
} }
} }

15
message/message-boot/src/main/java/com/njcn/message/produce/template/DeviceRebootMessageTemplate.java
View File

@@ -7,13 +7,10 @@ import com.njcn.message.message.AskRealDataMessage;
import com.njcn.message.message.DeviceRebootMessage; import com.njcn.message.message.DeviceRebootMessage;
import com.njcn.middle.rocket.domain.BaseMessage; import com.njcn.middle.rocket.domain.BaseMessage;
import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate; import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate;
import com.njcn.mq.core.MqTemplate;
import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/** /**
* 类的介绍： * 类的介绍：
* *
@@ -22,12 +19,14 @@ import javax.annotation.Resource;
* @createTime 2023/8/11 15:28 * @createTime 2023/8/11 15:28
*/ */
@Component @Component
public class DeviceRebootMessageTemplate { public class DeviceRebootMessageTemplate extends RocketMQEnhanceTemplate {
@Resource public DeviceRebootMessageTemplate(RocketMQTemplate template) {
private MqTemplate mqTemplate; super(template);
public void sendMember(BaseMessage baseMessage,String nodeId) { }
public SendResult sendMember(BaseMessage baseMessage,String nodeId) {
baseMessage.setSource(BusinessResource.WEB_RESOURCE); baseMessage.setSource(BusinessResource.WEB_RESOURCE);
mqTemplate.send(BusinessTopic.CONTROL_TOPIC,nodeId, baseMessage); return send(BusinessTopic.CONTROL_TOPIC,nodeId, baseMessage);
} }
} }

15
message/message-boot/src/main/java/com/njcn/message/produce/template/ProcessRebootMessageTemplate.java
View File

@@ -7,13 +7,10 @@ import com.njcn.message.message.AskRealDataMessage;
import com.njcn.message.message.ProcessRebootMessage; import com.njcn.message.message.ProcessRebootMessage;
import com.njcn.middle.rocket.domain.BaseMessage; import com.njcn.middle.rocket.domain.BaseMessage;
import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate; import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate;
import com.njcn.mq.core.MqTemplate;
import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/** /**
* 类的介绍： * 类的介绍：
* *
@@ -22,14 +19,16 @@ import javax.annotation.Resource;
* @createTime 2023/8/11 15:28 * @createTime 2023/8/11 15:28
*/ */
@Component @Component
public class ProcessRebootMessageTemplate { public class ProcessRebootMessageTemplate extends RocketMQEnhanceTemplate {
@Resource public ProcessRebootMessageTemplate(RocketMQTemplate template) {
private MqTemplate mqTemplate; super(template);
public void sendMember(BaseMessage baseMessage,String nodeId) { }
public SendResult sendMember(BaseMessage baseMessage,String nodeId) {
baseMessage.setSource(BusinessResource.WEB_RESOURCE); baseMessage.setSource(BusinessResource.WEB_RESOURCE);
ProcessRebootMessage dto = JSON.parseObject(baseMessage.getMessageBody(), ProcessRebootMessage.class); ProcessRebootMessage dto = JSON.parseObject(baseMessage.getMessageBody(), ProcessRebootMessage.class);
baseMessage.setKey(dto.getIndex()+""); baseMessage.setKey(dto.getIndex()+"");
mqTemplate.send(BusinessTopic.PROCESS_TOPIC,nodeId, baseMessage); return send(BusinessTopic.PROCESS_TOPIC,nodeId, baseMessage);
} }
} }

15
message/message-boot/src/main/java/com/njcn/message/produce/template/RecallMessaggeTemplate.java
View File

@@ -5,13 +5,10 @@ import com.njcn.message.constant.BusinessResource;
import com.njcn.message.constant.BusinessTopic; import com.njcn.message.constant.BusinessTopic;
import com.njcn.middle.rocket.domain.BaseMessage; import com.njcn.middle.rocket.domain.BaseMessage;
import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate; import com.njcn.middle.rocket.template.RocketMQEnhanceTemplate;
import com.njcn.mq.core.MqTemplate;
import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/** /**
* Description: * Description:
* Date: 2024/12/13 15:15【需求编号】 * Date: 2024/12/13 15:15【需求编号】
@@ -20,11 +17,13 @@ import javax.annotation.Resource;
* @version V1.0.0 * @version V1.0.0
*/ */
@Component @Component
public class RecallMessaggeTemplate { public class RecallMessaggeTemplate extends RocketMQEnhanceTemplate {
@Resource public RecallMessaggeTemplate(RocketMQTemplate template) {
private MqTemplate mqTemplate; super(template);
public void sendMember(BaseMessage recallMessage,String nodeId) { }
public SendResult sendMember(BaseMessage recallMessage,String nodeId) {
recallMessage.setSource(BusinessResource.WEB_RESOURCE); recallMessage.setSource(BusinessResource.WEB_RESOURCE);
mqTemplate.send(BusinessTopic.RECALL_TOPIC,nodeId , recallMessage); return send(BusinessTopic.RECALL_TOPIC,nodeId , recallMessage);
} }
} }

2
message/message-boot/src/main/resources/bootstrap.yml
View File

@@ -1,5 +1,3 @@
spring: spring:
profiles: profiles:
active: @spring.profiles.active@ active: @spring.profiles.active@
mq:
type: redis-stream

420
message/message-boot/src/test/java/com/njcn/message/mq/MqSliceWiringTest.java
View File

@@ -1,210 +1,210 @@
//package com.njcn.message.mq; package com.njcn.message.mq;
//
//import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
//import com.njcn.message.message.RecallMessage; import com.njcn.message.message.RecallMessage;
//import com.njcn.message.messagedto.MessageDataDTO; import com.njcn.message.messagedto.MessageDataDTO;
//import com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration; import com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration;
//import com.njcn.mq.autoconfig.MqCoreAutoConfiguration; import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
//import com.njcn.mq.container.MqListenerRegistry; import com.njcn.mq.container.MqListenerRegistry;
//import com.njcn.mq.core.MqTemplate; import com.njcn.mq.core.MqTemplate;
//import com.njcn.mq.driver.redis.RedisStreamMqDriver; import com.njcn.mq.driver.redis.RedisStreamMqDriver;
//import com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration; import com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration;
//import com.njcn.mq.driver.rocketmq.RocketMqDriver; import com.njcn.mq.driver.rocketmq.RocketMqDriver;
//import com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration; import com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration;
//import com.njcn.mq.spi.MqDriver; import com.njcn.mq.spi.MqDriver;
//import com.njcn.stat.api.MessAnalysisFeignClient; import com.njcn.stat.api.MessAnalysisFeignClient;
//import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
//import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.apache.rocketmq.spring.core.RocketMQTemplate;
//import org.junit.jupiter.api.Assumptions; import org.junit.jupiter.api.Assumptions;
//import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
//import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.autoconfigure.AutoConfigurations;
//import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration; import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
//import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.boot.test.context.runner.ApplicationContextRunner;
//import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.stream.MapRecord;
//import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.ReadOffset;
//import org.springframework.data.redis.connection.stream.StreamOffset; import org.springframework.data.redis.connection.stream.StreamOffset;
//import org.springframework.data.redis.connection.stream.StreamReadOptions; import org.springframework.data.redis.connection.stream.StreamReadOptions;
//import org.springframework.data.redis.connection.stream.StreamRecords; import org.springframework.data.redis.connection.stream.StreamRecords;
//import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate;
//
//import java.net.InetSocketAddress; import java.net.InetSocketAddress;
//import java.net.Socket; import java.net.Socket;
//import java.util.HashMap; import java.util.HashMap;
//import java.util.List; import java.util.List;
//import java.util.Map; import java.util.Map;
//
//import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
//import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyList;
//import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
//import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.timeout;
//import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verify;
//
///** /**
// * MQ starter 切片接入验证测试（ApplicationContextRunner，不启动全应用）。 * MQ starter 切片接入验证测试（ApplicationContextRunner，不启动全应用）。
// * 真实 Redis 要求：localhost:6379（docker redis-stream-it）；仅上/下行收发用例需要， * 真实 Redis 要求：localhost:6379（docker redis-stream-it）；仅上/下行收发用例需要，
// * 纯装配用例不需要（Lettuce 懒连接，不触发实际连接）。 * 纯装配用例不需要（Lettuce 懒连接，不触发实际连接）。
// * *
// * <p>装配一律用 {@code withConfiguration(AutoConfigurations.of(...))}，让 Spring 按 * <p>装配一律用 {@code withConfiguration(AutoConfigurations.of(...))}，让 Spring 按
// * <b>真实 autoconfig 顺序</b>（AutoConfigurationSorter）处理，与生产环境一致。以此 * <b>真实 autoconfig 顺序</b>（AutoConfigurationSorter）处理，与生产环境一致。以此
// * 验证 {@link MqCoreAutoConfiguration} 的 {@code @ConditionalOnBean(MqDriver)} 能在 * 验证 {@link MqCoreAutoConfiguration} 的 {@code @ConditionalOnBean(MqDriver)} 能在
// * 真实顺序下拿到 driver bean —— 这依赖 driver autoconfig 上的 * 真实顺序下拿到 driver bean —— 这依赖 driver autoconfig 上的
// * {@code @AutoConfigureBefore(MqCoreAutoConfiguration.class)}。若缺该保障，按类名字母序 * {@code @AutoConfigureBefore(MqCoreAutoConfiguration.class)}。若缺该保障，按类名字母序
// * {@code com.njcn.mq.autoconfig.*} 先于 {@code com.njcn.mq.driver.*}，core 会在 driver * {@code com.njcn.mq.autoconfig.*} 先于 {@code com.njcn.mq.driver.*}，core 会在 driver
// * bean 注册前求值条件失败，导致 MqTemplate / MqListenerRegistry 不装配。 * bean 注册前求值条件失败，导致 MqTemplate / MqListenerRegistry 不装配。
// */ */
//public class MqSliceWiringTest { public class MqSliceWiringTest {
//
// private static boolean redisAvailable() { private static boolean redisAvailable() {
// try { try {
// Socket s = new Socket(); Socket s = new Socket();
// s.connect(new InetSocketAddress("localhost", 6379), 500); s.connect(new InetSocketAddress("localhost", 6379), 500);
// s.close(); s.close();
// return true; return true;
// } catch (Exception e) { } catch (Exception e) {
// return false; return false;
// } }
// } }
//
// /** redis-stream 模式：真实 autoconfig 顺序装 5 个 autoconfig + 业务 bean（上/下行用）。 */ /** redis-stream 模式：真实 autoconfig 顺序装 5 个 autoconfig + 业务 bean（上/下行用）。 */
// private ApplicationContextRunner redisStreamRunner() { private ApplicationContextRunner redisStreamRunner() {
// return new ApplicationContextRunner() return new ApplicationContextRunner()
// .withConfiguration(AutoConfigurations.of( .withConfiguration(AutoConfigurations.of(
// RedisAutoConfiguration.class, RedisAutoConfiguration.class,
// RedisStreamAutoConfiguration.class, RedisStreamAutoConfiguration.class,
// RedisStreamMqDriverAutoConfiguration.class, RedisStreamMqDriverAutoConfiguration.class,
// RocketMqDriverAutoConfiguration.class, RocketMqDriverAutoConfiguration.class,
// MqCoreAutoConfiguration.class)) MqCoreAutoConfiguration.class))
// .withUserConfiguration(FrontDataMqListener.class, RecallMqSender.class) .withUserConfiguration(FrontDataMqListener.class, RecallMqSender.class)
// .withBean(MessAnalysisFeignClient.class, .withBean(MessAnalysisFeignClient.class,
// () -> mock(MessAnalysisFeignClient.class)) () -> mock(MessAnalysisFeignClient.class))
// .withPropertyValues( .withPropertyValues(
// "mq.type=redis-stream", "mq.type=redis-stream",
// "spring.redis.host=localhost", "spring.redis.host=localhost",
// "spring.redis.port=6379"); "spring.redis.port=6379");
// } }
//
// // ------------------------------------------------------------------ // ------------------------------------------------------------------
// // 1. 装配：mq.type 二选一 + core bean 在真实 autoconfig 顺序下装配 // 1. 装配：mq.type 二选一 + core bean 在真实 autoconfig 顺序下装配
// // ------------------------------------------------------------------ // ------------------------------------------------------------------
//
// @Test @Test
// void testRedisStreamDriverSelected() { void testRedisStreamDriverSelected() {
// // 纯装配检查，不依赖真实 Redis（Lettuce 懒连接）。 // 纯装配检查，不依赖真实 Redis（Lettuce 懒连接）。
// new ApplicationContextRunner() new ApplicationContextRunner()
// .withConfiguration(AutoConfigurations.of( .withConfiguration(AutoConfigurations.of(
// RedisAutoConfiguration.class, RedisAutoConfiguration.class,
// RedisStreamAutoConfiguration.class, RedisStreamAutoConfiguration.class,
// RedisStreamMqDriverAutoConfiguration.class, RedisStreamMqDriverAutoConfiguration.class,
// RocketMqDriverAutoConfiguration.class, RocketMqDriverAutoConfiguration.class,
// MqCoreAutoConfiguration.class)) MqCoreAutoConfiguration.class))
// .withPropertyValues( .withPropertyValues(
// "mq.type=redis-stream", "mq.type=redis-stream",
// "spring.redis.host=localhost", "spring.redis.host=localhost",
// "spring.redis.port=6379") "spring.redis.port=6379")
// .run(ctx -> { .run(ctx -> {
// assertThat(ctx).hasSingleBean(MqDriver.class); assertThat(ctx).hasSingleBean(MqDriver.class);
// assertThat(ctx.getBean(MqDriver.class)).isInstanceOf(RedisStreamMqDriver.class); assertThat(ctx.getBean(MqDriver.class)).isInstanceOf(RedisStreamMqDriver.class);
// // 真实顺序下 core 的 @ConditionalOnBean(MqDriver) 必须命中 // 真实顺序下 core 的 @ConditionalOnBean(MqDriver) 必须命中
// assertThat(ctx).hasSingleBean(MqTemplate.class); assertThat(ctx).hasSingleBean(MqTemplate.class);
// assertThat(ctx).hasSingleBean(MqListenerRegistry.class); assertThat(ctx).hasSingleBean(MqListenerRegistry.class);
// }); });
// } }
//
// @Test @Test
// void testRocketMqDriverSelected() { void testRocketMqDriverSelected() {
// // rocket 测试不依赖真实 Redis；不装 RedisAutoConfiguration，redis driver 因 mq.type 不匹配而 inactive。 // rocket 测试不依赖真实 Redis；不装 RedisAutoConfiguration，redis driver 因 mq.type 不匹配而 inactive。
// new ApplicationContextRunner() new ApplicationContextRunner()
// .withConfiguration(AutoConfigurations.of( .withConfiguration(AutoConfigurations.of(
// RedisStreamMqDriverAutoConfiguration.class, RedisStreamMqDriverAutoConfiguration.class,
// RocketMqDriverAutoConfiguration.class, RocketMqDriverAutoConfiguration.class,
// MqCoreAutoConfiguration.class)) MqCoreAutoConfiguration.class))
// .withBean(RocketMQTemplate.class, () -> mock(RocketMQTemplate.class)) .withBean(RocketMQTemplate.class, () -> mock(RocketMQTemplate.class))
// .withBean(RocketMQProperties.class, RocketMQProperties::new) .withBean(RocketMQProperties.class, RocketMQProperties::new)
// .withPropertyValues("mq.type=rocketmq") .withPropertyValues("mq.type=rocketmq")
// .run(ctx -> { .run(ctx -> {
// assertThat(ctx).hasSingleBean(MqDriver.class); assertThat(ctx).hasSingleBean(MqDriver.class);
// assertThat(ctx.getBean(MqDriver.class)).isInstanceOf(RocketMqDriver.class); assertThat(ctx.getBean(MqDriver.class)).isInstanceOf(RocketMqDriver.class);
// assertThat(ctx).hasSingleBean(MqTemplate.class); assertThat(ctx).hasSingleBean(MqTemplate.class);
// assertThat(ctx).hasSingleBean(MqListenerRegistry.class); assertThat(ctx).hasSingleBean(MqListenerRegistry.class);
// }); });
// } }
//
// // ------------------------------------------------------------------ // ------------------------------------------------------------------
// // 2. 上行（需要真实 Redis） // 2. 上行（需要真实 Redis）
// // 向 LN_Topic stream XADD 一条消息，验证 FrontDataMqListener 触发了 // 向 LN_Topic stream XADD 一条消息，验证 FrontDataMqListener 触发了
// // messAnalysisFeignClient.analysis(...) // messAnalysisFeignClient.analysis(...)
// // ------------------------------------------------------------------ // ------------------------------------------------------------------
//
// @Test @Test
// void testUpstreamFrontData() { void testUpstreamFrontData() {
// Assumptions.assumeTrue(redisAvailable(), "Redis not available at localhost:6379, skipping"); Assumptions.assumeTrue(redisAvailable(), "Redis not available at localhost:6379, skipping");
// redisStreamRunner().run(ctx -> { redisStreamRunner().run(ctx -> {
// StringRedisTemplate redis = ctx.getBean(StringRedisTemplate.class); StringRedisTemplate redis = ctx.getBean(StringRedisTemplate.class);
// MessAnalysisFeignClient mockClient = ctx.getBean(MessAnalysisFeignClient.class); MessAnalysisFeignClient mockClient = ctx.getBean(MessAnalysisFeignClient.class);
//
// // 构造 RedisStreamMqDriver 消费时期望的字段格式 // 构造 RedisStreamMqDriver 消费时期望的字段格式
// // enc=plain → body 即为原始 JSON（参见 MessageCodec.encodeBody(json, false)） // enc=plain → body 即为原始 JSON（参见 MessageCodec.encodeBody(json, false)）
// // stream key = StreamKeyBuilder.buildStream("LN_Topic") = "LN_Topic" // stream key = StreamKeyBuilder.buildStream("LN_Topic") = "LN_Topic"
// // （默认 envIsolation=false，无 env 前缀） // （默认 envIsolation=false，无 env 前缀）
// MessageDataDTO dto = new MessageDataDTO(); MessageDataDTO dto = new MessageDataDTO();
// dto.setDataType(1); dto.setDataType(1);
// String json = JSON.toJSONString(dto); String json = JSON.toJSONString(dto);
//
// Map<String, String> fields = new HashMap<String, String>(); Map<String, String> fields = new HashMap<String, String>();
// fields.put("key", "test-key-upstream-1"); fields.put("key", "test-key-upstream-1");
// fields.put("tag", ""); fields.put("tag", "");
// fields.put("enc", "plain"); fields.put("enc", "plain");
// fields.put("body", json); fields.put("body", json);
// fields.put("ts", String.valueOf(System.currentTimeMillis())); fields.put("ts", String.valueOf(System.currentTimeMillis()));
//
// // XADD 到 LN_Topic（MqListenerRegistry.start() 已在 context 启动时订阅该 stream） // XADD 到 LN_Topic（MqListenerRegistry.start() 已在 context 启动时订阅该 stream）
// redis.opsForStream().add( redis.opsForStream().add(
// StreamRecords.newRecord().in("LN_Topic").ofMap(fields) StreamRecords.newRecord().in("LN_Topic").ofMap(fields)
// ); );
//
// // 等待消费线程处理（blockMs 默认 2000ms，timeout=5000ms 足够） // 等待消费线程处理（blockMs 默认 2000ms，timeout=5000ms 足够）
// verify(mockClient, timeout(5000)).analysis(anyList()); verify(mockClient, timeout(5000)).analysis(anyList());
// }); });
// } }
//
// // ------------------------------------------------------------------ // ------------------------------------------------------------------
// // 3. 下行（需要真实 Redis） // 3. 下行（需要真实 Redis）
// // 通过 RecallMqSender 发送 RecallMessage，验证 stream 内容正确 // 通过 RecallMqSender 发送 RecallMessage，验证 stream 内容正确
// // ------------------------------------------------------------------ // ------------------------------------------------------------------
//
// @Test @Test
// void testDownstreamRecall() { void testDownstreamRecall() {
// Assumptions.assumeTrue(redisAvailable(), "Redis not available at localhost:6379, skipping"); Assumptions.assumeTrue(redisAvailable(), "Redis not available at localhost:6379, skipping");
// redisStreamRunner().run(ctx -> { redisStreamRunner().run(ctx -> {
// StringRedisTemplate redis = ctx.getBean(StringRedisTemplate.class); StringRedisTemplate redis = ctx.getBean(StringRedisTemplate.class);
// RecallMqSender sender = ctx.getBean(RecallMqSender.class); RecallMqSender sender = ctx.getBean(RecallMqSender.class);
//
// // 清理，确保只有本次发送的消息 // 清理，确保只有本次发送的消息
// redis.delete("n1_recall_Topic"); redis.delete("n1_recall_Topic");
//
// RecallMessage msg = new RecallMessage(); RecallMessage msg = new RecallMessage();
// msg.setGuid("g1"); msg.setGuid("g1");
// // MqTemplate.send("n1_recall_Topic", msg) → RedisStreamMqDriver XADD // MqTemplate.send("n1_recall_Topic", msg) → RedisStreamMqDriver XADD
// // stream key = StreamKeyBuilder.buildStream("n1_recall_Topic") = "n1_recall_Topic" // stream key = StreamKeyBuilder.buildStream("n1_recall_Topic") = "n1_recall_Topic"
// sender.send(msg, "n1"); sender.send(msg, "n1");
//
// Long size = redis.opsForStream().size("n1_recall_Topic"); Long size = redis.opsForStream().size("n1_recall_Topic");
// assertThat(size).isGreaterThanOrEqualTo(1L); assertThat(size).isGreaterThanOrEqualTo(1L);
//
// @SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
// List<MapRecord<String, Object, Object>> records = List<MapRecord<String, Object, Object>> records =
// redis.opsForStream().read( redis.opsForStream().read(
// StreamReadOptions.empty().count(1L), StreamReadOptions.empty().count(1L),
// StreamOffset.create("n1_recall_Topic", ReadOffset.from("0-0")) StreamOffset.create("n1_recall_Topic", ReadOffset.from("0-0"))
// ); );
// assertThat(records).isNotEmpty(); assertThat(records).isNotEmpty();
//
// Map<Object, Object> firstFields = records.get(0).getValue(); Map<Object, Object> firstFields = records.get(0).getValue();
// // DefaultMqTemplate.send(topic, payload) 使用 encodeBody(json, false) → enc="plain" // DefaultMqTemplate.send(topic, payload) 使用 encodeBody(json, false) → enc="plain"
// // 因此 body 即为原始 JSON，无需 gzip 解压 // 因此 body 即为原始 JSON，无需 gzip 解压
// String body = (String) firstFields.get("body"); String body = (String) firstFields.get("body");
// RecallMessage result = JSON.parseObject(body, RecallMessage.class); RecallMessage result = JSON.parseObject(body, RecallMessage.class);
// assertThat(result.getGuid()).isEqualTo("g1"); assertThat(result.getGuid()).isEqualTo("g1");
// }); });
// } }
//} }

4
pom.xml
View File

@@ -34,8 +34,8 @@
<properties> <properties>
<spring.profiles.active>sjzx</spring.profiles.active> <spring.profiles.active>sjzx</spring.profiles.active>
<!--内网--> <!--内网-->
<middle.server.url>192.168.1.68</middle.server.url> <middle.server.url>192.168.1.65</middle.server.url>
<service.server.url>192.168.2.130</service.server.url> <service.server.url>192.168.1.65</service.server.url>
<docker.server.url>192.168.1.22</docker.server.url> <docker.server.url>192.168.1.22</docker.server.url>
<nacos.url>${middle.server.url}:18848</nacos.url> <nacos.url>${middle.server.url}:18848</nacos.url>
<nacos.namespace></nacos.namespace> <nacos.namespace></nacos.namespace>