From 72e63be140b603d12c89bd4f57d5663376ce941c Mon Sep 17 00:00:00 2001 From: hzj <826100833@qq.com> Date: Mon, 7 Apr 2025 11:24:11 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BA=A4=E9=85=8D=E7=BD=AE=E6=96=87?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../consumer/FrontDataConsumerTest.java | 173 ++++++++++++++++++ .../src/main/resources/bootstrap.yml | 2 +- 2 files changed, 174 insertions(+), 1 deletion(-) create mode 100644 message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumerTest.java diff --git a/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumerTest.java b/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumerTest.java new file mode 100644 index 0000000..b9a90ef --- /dev/null +++ b/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumerTest.java @@ -0,0 +1,173 @@ +package com.njcn.message.consumer; + +import com.alibaba.fastjson.JSONObject; +import com.njcn.message.constant.MessageStatus; +import com.njcn.message.messagedto.MessageDataDTO; +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.system.api.RocketMqLogFeignClient; +import com.njcn.system.pojo.po.RocketmqMsgErrorLog; +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.stereotype.Component; + +import javax.annotation.Resource; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; + + +/** + * 类的介绍： + * + * @author xuyang + * @version 1.0.0 + * @createTime 2023/8/11 15:32 + */ +@Component +@RocketMQMessageListener( + topic = "Test_Topic", + consumerGroup = "Test_consumer", + selectorExpression = "Test_Tag||Test_Keys", + consumeThreadNumber = 10, + enableMsgTrace = true +) +@Slf4j +public class FrontDataConsumerTest extends EnhanceConsumerMessageHandler implements RocketMQListener { + + @Autowired + private MessAnalysisFeignClient messAnalysisFeignClient; + + + + @Resource + private RedisUtil redisUtil; + + @Resource + private RocketMqLogFeignClient rocketMqLogFeignClient; + + + private List messageList = new ArrayList<>(1); + + @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(message.getKey(), MessageStatus.BEING_PROCESSED, 60L); + return false; + } + return true; + } + /** + * 消费成功，缓存到redis72小时，避免重复消费 + */ + @Override + protected void consumeSuccess(MessageDataDTO message) { + redisUtil.saveByKeyWithExpire(message.getKey(), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime()); + } + + + @Override + protected void handleMessage(MessageDataDTO message) { + synchronized (messageList) { + messageList.add(message); + if (messageList.size() >= 1) { + saveToDatabase(); + } + } + } + + + + + /** + * 发生异常时，进行错误信息入库保存 + * 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存 + */ + @Override + protected void saveExceptionMsgLog(MessageDataDTO message, String identity, Exception exception) { + redisUtil.saveByKeyWithExpire(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("处理120条消息所需时间------------"+(end-start)); + }catch (Exception e){{ + log.info(e.toString()); + } + }finally{ + messageList.clear(); + } + + } + + + + +} diff --git a/message/message-boot/src/main/resources/bootstrap.yml b/message/message-boot/src/main/resources/bootstrap.yml index 3898c94..1f93eda 100644 --- a/message/message-boot/src/main/resources/bootstrap.yml +++ b/message/message-boot/src/main/resources/bootstrap.yml @@ -29,7 +29,7 @@ spring: shared-configs: - data-id: share-config.yaml refresh: true - - data-Id: algorithm-config.yaml + - data-Id: data-platform.yaml refresh: true main: allow-bean-definition-overriding: true