From 92c64873e6a59bce3c59d826ea8dc84a3d936ba1 Mon Sep 17 00:00:00 2001 From: hzj <826100833@qq.com> Date: Mon, 7 Apr 2025 11:24:58 +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 | 346 +++++++++--------- 1 file changed, 173 insertions(+), 173 deletions(-) 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 index b9a90ef..2daacca 100644 --- 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 @@ -1,173 +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(); - } - - } - - - - -} +//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(); +// } +// +// } +// +// +// +// +//}