liuyupeng/saas-houduan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复部分设备注册时networklist为null导致的失败问题

Browse Source
This commit is contained in:
Your Name
2026-07-27 10:50:32 +08:00
parent c918523a7a
commit 368121b938
2 changed files with 14 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java
+12 -10
View File
@@ -2,6 +2,7 @@ package com.tongran.rocketmq.consumer;
import com.alibaba.fastjson.JSON;
import com.tongran.common.core.enums.MsgEnum;
import com.tongran.common.core.utils.ExceptionUtil;
import com.tongran.rocketmq.domain.DeviceMessage;
import com.tongran.rocketmq.enums.MessageCodeEnum;
import com.tongran.rocketmq.handler.DeviceMessageHandler;
@@ -41,7 +42,8 @@ public class RocketMsgListener implements MessageListenerConcurrently {
*/
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
try{
String msgId = "";
try{
//消息不等于空情况
if (!CollectionUtils.isEmpty(list)) {
//获取topic
@@ -49,11 +51,11 @@ public class RocketMsgListener implements MessageListenerConcurrently {
// 解析消息内容
// 明确指定UTF-8编码
String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
log.info("接受到的消息为：{}", body);
String tags = messageExt.getTags();
String topic = messageExt.getTopic();
String msgId = messageExt.getMsgId();
msgId = messageExt.getMsgId();
String keys = messageExt.getKeys();
log.info("接收到消息, msgId={}, body={}", msgId, body);
int reConsume = messageExt.getReconsumeTimes();
// 消息已经重试了3次，如果不需要再次消费，则返回成功
if (reConsume == 3) {
@@ -61,14 +63,14 @@ public class RocketMsgListener implements MessageListenerConcurrently {
DeviceMessage message = JSON.parseObject(body, DeviceMessage.class);
if(!message.getDataType().equals(MsgEnum.网络上报重试.getValue())){
// TODO 补偿信息
log.error("消息消费三次失败，消息内容：{}", body);
log.error("消息消费三次失败，msgId={}, body={}", body);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//根据业务返回是否正常
}
}
// 流量数据重试
if (reConsume == 6) {
// 补偿信息
log.error("流量数据重试消息消费6次失败，消息内容：{}", body);
log.error("流量数据重试消息消费六次失败，msgId={}, body={}", body);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//根据业务返回是否正常
}
if(MessageCodeEnum.TONGRAN_AGENT_UP.getCode().equals(topic)){
@@ -93,7 +95,7 @@ public class RocketMsgListener implements MessageListenerConcurrently {
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
} catch (Exception e) {
// 调用 handleException 方法处理异常并返回处理结果
return handleException(e);
return handleException(msgId, e);
}
}
@@ -103,14 +105,14 @@ public class RocketMsgListener implements MessageListenerConcurrently {
* @param e 捕获的异常
* @return 消息消费结果
*/
private static ConsumeConcurrentlyStatus handleException(final Exception e) {
private static ConsumeConcurrentlyStatus handleException(String msgId, final Exception e) {
Class exceptionClass = e.getClass();
if (exceptionClass.equals(UnsupportedEncodingException.class)) {
log.error(e.getMessage());
log.error("消息处理失败, msgId={}, error={}", msgId, e.getMessage());
} else if (exceptionClass.equals(ConsumeException.class)) {
log.error(e.getMessage());
log.error("消息处理失败, msgId={}, error={}", msgId, e.getMessage());
} else{
log.error(e.getMessage());
log.error("消息处理失败, msgId={}, error={} {}", msgId, e.getMessage(), ExceptionUtil.getExceptionMessage(e));
}
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java
+2 -2
View File
@@ -768,7 +768,7 @@ public class MessageHandler {
// 自动注册服务器信息
RmRegisterMsgRemote rmRegisterMsgRemote = new RmRegisterMsgRemote();
BeanUtils.copyProperties(registerMsg, rmRegisterMsgRemote);
int rows = remoteRevenueConfigService.innerAddRegist(rmRegisterMsgRemote, SecurityConstants.INNER).getData();
int rows = Optional.ofNullable(remoteRevenueConfigService.innerAddRegist(rmRegisterMsgRemote, SecurityConstants.INNER).getData()).orElse(0);
if(rows == 2){
// 注册成功，下发优先级为0的策略
rmMonitorPolicyService.issueDefaultPolicyByClientId(message.getClientId());
@@ -1963,7 +1963,7 @@ public class MessageHandler {
Date createTime = new Date(millis / 1000 * 1000);
List<NetworkInfo> networkInfoList = registerMsg.getNetworkInfo();
if (networkInfoList.isEmpty()) {
if (networkInfoList == null || networkInfoList.isEmpty()) {
return;
}