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; }