From 368121b93809cd61d4c1d7a428b235fadf522aac Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 27 Jul 2026 10:50:32 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E9=83=A8=E5=88=86=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E6=B3=A8=E5=86=8C=E6=97=B6networklist=E4=B8=BAnull?= =?UTF-8?q?=E5=AF=BC=E8=87=B4=E7=9A=84=E5=A4=B1=E8=B4=A5=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../rocketmq/consumer/RocketMsgListener.java | 22 ++++++++++--------- .../rocketmq/handler/MessageHandler.java | 4 ++-- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java index 5a118ac..668e474 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java @@ -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 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; } diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java index abdd104..6786399 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java @@ -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 networkInfoList = registerMsg.getNetworkInfo(); - if (networkInfoList.isEmpty()) { + if (networkInfoList == null || networkInfoList.isEmpty()) { return; }