devbox
运维做开发
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

SpringBoot优雅的封装不同研发环境下（环境隔离）RocketMq自动ack和手动ack

作者：运维做开发 | 2024-08-16 13:21:48

赞

踩

SpringBoot优雅的封装不同研发环境下（环境隔离）RocketMq自动ack和手动ack

1. RocketMq的maven依赖版本：

  1.      <dependency>
  2.            <groupId>org.apache.rocketmq</groupId>
  3.             <artifactId>rocketmq-spring-boot-starter</artifactId>
  4.             <version>2.3.0</version>
  5.      </dependency>

2.RocketMq的yml文件：

  1. # 自定义属性
  2. system:
  3.   environment:
  4.     # 隔离环境名称，拼接到topic后，xxx_topic_tianxin，默认空字符串
  5.     name: dev
  6.     # 启动隔离，会自动在topic上拼接激活的配置文件，达到自动隔离的效果
  7.     # 默认为true，配置类：EnvironmentIsolationConfig
  8.     isolation: true
  9. rocketmq:
  10.   # 多个NameServer，host:port;host:port，RocketMQProperties
  11.   nameServer: 你的NameServer
  12.   producer:
  13.     # 发o送同一类消息的设置为同一个grup，保证唯一
  14.     group: logistics_group
  15.     # 发送消息失败重试次数，默认2
  16.     retryTimesWhenSendFailed: 2
  17.     # 异步消息重试此处，默认2
  18.     retryTimesWhenSendAsyncFailed: 2
  19.     # 发送消息超时时间，默认3000
  20.     sendMessageTimeout: 10000
  21.     # 消息最大长度，默认1024 * 1024 * 4(默认4M)
  22.     maxMessageSize: 4096
  23.     # 压缩消息阈值，默认4k(1024 * 4)
  24.     compressMessageBodyThreshold: 4096
  25.     # 是否在内部发送失败时重试另一个broker，默认false
  26.     retryNextServer: false
  27.     # access-key
  28.     accessKey: 你的access-key
  29.     # secret-key
  30.     secretKey: 你的secret-key
  31.     # 是否启用消息跟踪，默认false
  32.     enableMsgTrace: false
  33.     # 消息跟踪主题的名称值。如果不进行配置，可以使用默认的跟踪主题名称
  34.     customizedTraceTopic: RMQ_SYS_TRACE_TOPIC
  35.   consumer:
  36.     # 指定消费组
  37.     group: logistics_group
  38.     #广播消费模式 CLUSTERING(集群消费)、BROADCASTING(广播消费)
  39.     messageModel: CLUSTERING
  40.     #设置消费超时时间(分钟)
  41.     consumeTimeout: 1
  42.     # 最大重试次数，默认16
  43.     maxReconsumeTimes: 3
  44.     # 其他配置参考属性类

3 IsolationConfig读取yml文件配置

  1. package com.logistics.common.rocketMq.config;
  2. import lombok.Data;
  3. import org.springframework.beans.factory.annotation.Value;
  4. import org.springframework.context.annotation.Configuration;
  5.  
  6. /**
  7.  * @author: 吴顺杰
  8.  * @create: 2024-06-18 10:01
  9.  * @Description:
  10.  * RocketMQ多环境隔离配置
  11.  * 原理：对于每个配置的Bean在实例化前，拿到Bean的监听器注解把group或者topic改掉
  12.  */
  13. @Configuration
  14. @Data
  15. public class IsolationConfig {
  16.     @Value("${system.environment.isolation:true}")
  17.     private boolean enabledIsolation;
  18.     @Value("${system.environment.name:''}")
  19.     private String environmentName;
  20.     @Value("${rocketmq.nameServer:''}")
  21.     private String nameServer;
  22.     @Value("${rocketmq.consumer.group:''}")
  23.     private String group;
  24.     @Value("${rocketmq.consumer.messageModel:''}")
  25.     private String messageModel;
  26.     @Value("${rocketmq.consumer.consumeTimeout:''}")
  27.     private int consumeTimeout;
  28.     @Value("${rocketmq.consumer.maxReconsumeTimes:''}")
  29.     private int maxReconsumeTimes;
  30. }

4.RocketMQ序列化器处理RocketMqConfig文件

主要是为了解决RocketMQ Jackson不支持Java时间类型配置

  1. package com.logistics.common.rocketMq.config;
  2.  
  3. import com.fasterxml.jackson.databind.ObjectMapper;
  4. import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
  5. import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  8. import org.springframework.context.annotation.Primary;
  9. import org.springframework.messaging.converter.CompositeMessageConverter;
  10. import org.springframework.messaging.converter.MappingJackson2MessageConverter;
  11. import org.springframework.messaging.converter.MessageConverter;
  12.  
  13. import java.util.List;
  14.  
  15. /**
  16.  * RocketMQ序列化器处理
  17.  *
  18.  * @author 吴顺杰
  19.  * @since 2024/8/04
  20.  */
  21. @Configuration
  22. public class RocketMqConfig {
  23.  
  24.     /**
  25.      * 解决RocketMQ Jackson不支持Java时间类型配置
  26.      */
  27.     @Bean
  28.     @Primary
  29.     public RocketMQMessageConverter createRocketMQMessageConverter() {
  30.         RocketMQMessageConverter converter = new RocketMQMessageConverter();
  31.         CompositeMessageConverter compositeMessageConverter = (CompositeMessageConverter) converter.getMessageConverter();
  32.         List<MessageConverter> messageConverterList = compositeMessageConverter.getConverters();
  33.         for (MessageConverter messageConverter : messageConverterList) {
  34.             if (messageConverter instanceof MappingJackson2MessageConverter) {
  35.                 MappingJackson2MessageConverter jackson2MessageConverter = (MappingJackson2MessageConverter) messageConverter;
  36.                 ObjectMapper objectMapper = jackson2MessageConverter.getObjectMapper();
  37.                 // 增加Java8时间模块支持，实体类可以传递LocalDate/LocalDateTime
  38.                 objectMapper.registerModules(new JavaTimeModule());
  39.             }
  40.         }
  41.         return converter;
  42.     }
  43. }

JSON工具类封装

  1. package com.logistics.common.rocketMq.utils;
  2.  
  3. import com.alibaba.fastjson.JSONObject;
  4.  
  5. /**
  6.  * JSON工具类
  7.  * 像工具类这种，建议一定要二次封装，避免出现漏洞时可以快速替换
  8.  *
  9.  * @author 吴顺杰
  10.  * @since 2024/6/16
  11.  */
  12. public class JsonUtil {
  13.     private JsonUtil() {}
  14.  
  15.     public static String toJson(Object value) {
  16.         return JSONObject.toJSONString(value);
  17.     }
  18.  
  19.     public static <T> T toObject(String jsonStr, Class<T> clazz) {
  20.         return JSONObject.parseObject(jsonStr, clazz);
  21.     }
  22. }

5.rocketMq生产者封装

调度任务生产者：LogisticsAddDispatchMqProducer

  1. package com.logistics.business.rocketMq.producer;
  2.  
  3. import com.logistics.common.exception.base.BaseException;
  4. import com.logistics.common.rocketMq.constant.AddDispatchMqContant;
  5. import com.logistics.common.rocketMq.domain.AddDispatchMqMessage;
  6. import com.logistics.common.rocketMq.template.RocketMqTemplate;
  7. import com.logistics.common.utils.spring.SpringUtils;
  8. import lombok.extern.slf4j.Slf4j;
  9.  
  10. /**
  11.  * 新增调度任务生产者MQ队列
  12.  */
  13. @Slf4j
  14. public class LogisticsAddDispatchMqProducer {
  15.     private static RocketMqTemplate rocketMqTemplate = SpringUtils.getBean(RocketMqTemplate.class);
  16.  
  17.     public static void sendAddDispatchMqMessage(AddDispatchMqMessage message) {
  18.         log.info("新增调度任务成功发送新增调度消息MQ，内容: {}", message);
  19.         try {
  20.             rocketMqTemplate.asyncSend(AddDispatchMqContant.ADD_DISPATCH_TOPIC, AddDispatchMqContant.ADD_DISPATCH_TAG, message);
  21.         } catch (Exception e) {
  22.             log.error("新增调度任务成功发送新增调度消息MQ失败，内容: {}", message, e);
  23.             throw new BaseException("新增调度任务成功发送新增调度消息MQ失败，请联系管理员");
  24.         }
  25.     }
  26. }

RocketMQ模板类

  1. package com.logistics.common.rocketMq.template;
  2.  
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.logistics.common.rocketMq.config.IsolationConfig;
  5. import com.logistics.common.rocketMq.constant.RocketMqSysConstant;
  6. import com.logistics.common.rocketMq.domain.BaseMqMessage;
  7. import com.logistics.common.rocketMq.utils.JsonUtil;
  8. import lombok.extern.slf4j.Slf4j;
  9. import org.apache.rocketmq.client.producer.SendCallback;
  10. import org.apache.rocketmq.client.producer.SendResult;
  11. import org.apache.rocketmq.client.producer.SendStatus;
  12. import org.apache.rocketmq.spring.core.RocketMQTemplate;
  13. import org.apache.rocketmq.spring.support.RocketMQHeaders;
  14. import org.slf4j.Logger;
  15. import org.slf4j.LoggerFactory;
  16. import org.springframework.messaging.Message;
  17. import org.springframework.messaging.support.MessageBuilder;
  18. import org.springframework.stereotype.Component;
  19. import org.springframework.util.StringUtils;
  20.  
  21. import javax.annotation.Resource;
  22.  
  23. /**
  24.  * RocketMQ模板类
  25.  *
  26.  */
  27. @Component
  28. @Slf4j
  29. public class RocketMqTemplate {
  30.     private static final Logger LOGGER = LoggerFactory.getLogger(RocketMqTemplate.class);
  31.     @Resource(name = "rocketMQTemplate")
  32.     private RocketMQTemplate template;
  33.     @Resource
  34.     private IsolationConfig isolationConfig;
  35.  
  36.  
  37.     /**
  38.      * 获取模板，如果封装的方法不够提供原生的使用方式
  39.      */
  40.     public RocketMQTemplate getTemplate() {
  41.         return template;
  42.     }
  43.  
  44.     /**
  45.      * 构建目的地
  46.      */
  47.     public String buildDestination(String topic, String tag) {
  48.         return topic + RocketMqSysConstant.DELIMITER + tag;
  49.     }
  50.  
  51.     /**
  52.      * 发送同步消息
  53.      */
  54.     public <T extends BaseMqMessage> SendResult syncSend(String topic, String tag, T message) {
  55.         // 开启消息隔离情况下获取隔离配置，隔离topic，根据自己的需求隔离group或者tag
  56.         if (isolationConfig.isEnabledIsolation() && StringUtils.hasText(isolationConfig.getEnvironmentName())) {
  57.             topic = topic + "_" + isolationConfig.getEnvironmentName();
  58.         }
  59.         // 设置业务键，此处根据公共的参数进行处理
  60.         Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
  61.         String buildDestination = buildDestination(topic, tag);
  62.         SendResult sendResult = template.syncSend(buildDestination, sendMessage);
  63.         // 此处为了方便查看给日志转了json，根据选择选择日志记录方式，例如ELK采集
  64.         LOGGER.info("[{}]同步消息[{}]发送结果[{}]", buildDestination, JsonUtil.toJson(message), JSONObject.toJSON(sendResult));
  65.         return sendResult;
  66.     }
  67.  
  68.  
  69.     /**
  70.      * 发送异步消息
  71.      *
  72.      * @param topic
  73.      * @param tag
  74.      * @param message
  75.      * @param <T>
  76.      */
  77.     public <T extends BaseMqMessage> void asyncSend(String topic, String tag, T message) {
  78.         // 开启消息隔离情况下获取隔离配置，隔离topic，根据自己的需求隔离group或者tag
  79.         if (isolationConfig.isEnabledIsolation() && StringUtils.hasText(isolationConfig.getEnvironmentName())) {
  80.             topic = topic + "_" + isolationConfig.getEnvironmentName();
  81.         }
  82.         // 设置业务键，此处根据公共的参数进行处理
  83.         Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
  84.         String buildDestination = buildDestination(topic, tag);
  85.         template.asyncSend(buildDestination, sendMessage, new SendCallback() {
  86.             @Override
  87.             public void onSuccess(SendResult sendResult) {
  88.                 LOGGER.info("[{}]MQ异步消息[{}]发送成功结果[{}]", buildDestination, JsonUtil.toJson(message), JSONObject.toJSON(sendResult));
  89.                 if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
  90.                     //可以存入数据库做处理
  91.                     log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());
  92.                 }
  93.             }
  94.  
  95.             @Override
  96.             public void onException(Throwable throwable) {
  97.                 LOGGER.info("[{}]MQ异步消息[{}]发送失败结果[{}]", buildDestination, JsonUtil.toJson(message), JSONObject.toJSON(throwable.getMessage()));
  98.                 //可以存入数据库做处理
  99.             }
  100.         });
  101.     }
  102.  
  103.     /**
  104.      * 发送延迟消息
  105.      *
  106.      * @param message
  107.      * @param delayLevel
  108.      * @param <T>
  109.      * @return
  110.      */
  111.     public <T extends BaseMqMessage> SendResult syncDelaySend(String topic, String tag, T message, int delayLevel) {
  112.         // 开启消息隔离情况下获取隔离配置，隔离topic，根据自己的需求隔离group或者tag
  113.         if (isolationConfig.isEnabledIsolation() && StringUtils.hasText(isolationConfig.getEnvironmentName())) {
  114.             topic = topic + "_" + isolationConfig.getEnvironmentName();
  115.         }
  116.         Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getKey()).build();
  117.         String destination = buildDestination(topic, tag);
  118.         SendResult sendResult = template.syncSend(destination, sendMessage, 3000, delayLevel);
  119.         LOGGER.info("[{}]延迟等级[{}]消息[{}]发送结果[{}]", destination, delayLevel, JsonUtil.toJson(message), JsonUtil.toJson(sendResult));
  120.         return sendResult;
  121.     }
  122.  
  123. }

AddDispatchMqMessage消息实体

  1. package com.logistics.common.rocketMq.domain;
  2.  
  3. import lombok.Builder;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6.  
  7.  
  8. /**
  9.  * 新增调度任务发送队列消息体
  10.  */
  11. @Data
  12. @Builder
  13. @NoArgsConstructor
  14. public class AddDispatchMqMessage extends BaseMqMessage {
  15.     /**
  16.      * 调度id
  17.      */
  18.     private Long dispatchId;
  19.  
  20.     public AddDispatchMqMessage(Long dispatchId) {
  21.         this.dispatchId = dispatchId;
  22.     }
  23. }

AddDispatchMqContant类

  1. package com.logistics.common.rocketMq.constant;
  2.  
  3. /**
  4.  * 新增调度任务MQ队列
  5.  */
  6. public class AddDispatchMqContant {
  7.     /**
  8.      * 消费主题
  9.      */
  10.     public static final String ADD_DISPATCH_TOPIC = "add_dispatch_topic";
  11.  
  12.     /**
  13.      * 消费标签
  14.      */
  15.     public static final String ADD_DISPATCH_TAG = "add_dispatch_tag";
  16.  
  17.     /**
  18.      * 消费组
  19.      */
  20.     public static final String ADD_DISPATCH_GROUP = "add_dispatch_group";
  21. }

6.rocketMq消费者封装

ACK简介
在实际使用RocketMQ的时候我们并不能保证每次发送的消息都刚好能被消费者一次性正常消费成功，可能会存在需要多次消费才能成功或者一直消费失败的情况，那作为发送者该做如何处理呢？

RocketMQ提供了ack机制，以保证消息能够被正常消费。发送者为了保证消息肯定消费成功，只有使用方明确表示消费成功，RocketMQ才会认为消息消费成功。中途断电，抛出异常等都不会认为成功——即都会重新投递。

1.手动ack封装

新增增调度任务消费者启动监听类

  1. package com.logistics.business.rocketMq.listener;
  2.  
  3. import com.logistics.business.rocketMq.comsumer.LogisticsAddDispatchMqComsumer;
  4. import com.logistics.common.rocketMq.config.IsolationConfig;
  5. import com.logistics.common.rocketMq.constant.AddDispatchMqContant;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  8. import org.apache.rocketmq.client.exception.MQClientException;
  9. import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
  10. import org.springframework.stereotype.Component;
  11. import org.springframework.util.StringUtils;
  12.  
  13. import javax.annotation.PostConstruct;
  14. import javax.annotation.Resource;
  15.  
  16.  
  17. /**
  18.  * 新增调度任务消费者启动监听类
  19.  */
  20. @Component
  21. @Slf4j
  22. public class RetryLogisticsAddDispatchListener {
  23.     @Resource
  24.     private IsolationConfig isolationConfig;
  25.  
  26.     private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
  27.  
  28.     @PostConstruct
  29.     public void start() {
  30.         try {
  31.             //启动环境隔离
  32.             String topic = AddDispatchMqContant.ADD_DISPATCH_TOPIC;
  33.             if (isolationConfig.isEnabledIsolation() && StringUtils.hasText(isolationConfig.getEnvironmentName())) {
  34.                 consumer.setConsumerGroup(AddDispatchMqContant.ADD_DISPATCH_GROUP + "_" + isolationConfig.getEnvironmentName());
  35.                 topic = topic + "_" + isolationConfig.getEnvironmentName();
  36.             } else {
  37.                 consumer.setConsumerGroup(AddDispatchMqContant.ADD_DISPATCH_GROUP);
  38.             }
  39.  
  40.             consumer.setNamesrvAddr(isolationConfig.getNameServer());
  41.             //设置集群消费模式
  42.             consumer.setMessageModel(MessageModel.valueOf(isolationConfig.getMessageModel()));
  43.             //设置消费超时时间(分钟)
  44.             consumer.setConsumeTimeout(isolationConfig.getConsumeTimeout());
  45.             //最大重试次数
  46.             consumer.setMaxReconsumeTimes(isolationConfig.getMaxReconsumeTimes());
  47.             //订阅主题
  48.             consumer.subscribe(topic, AddDispatchMqContant.ADD_DISPATCH_TAG);
  49.             //注册消息监听器
  50.             consumer.registerMessageListener(new LogisticsAddDispatchMqComsumer());
  51.             //启动消费端
  52.             consumer.start();
  53.             log.info("新增调度任务消费者MQ监听队列启动成功");
  54.         } catch (MQClientException e) {
  55.             e.printStackTrace();
  56.         }
  57.     }
  58. }

LogisticsAddDispatchMqComsumer注册消息监听器

  1. package com.logistics.business.rocketMq.comsumer;
  2.  
  3. import com.logistics.common.rocketMq.domain.AddDispatchMqMessage;
  4. import com.logistics.common.rocketMq.utils.JsonUtil;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.apache.commons.collections.CollectionUtils;
  7. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  8. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  9. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  10. import org.apache.rocketmq.common.message.MessageExt;
  11.  
  12. import java.nio.charset.StandardCharsets;
  13. import java.util.List;
  14.  
  15. /**
  16.  * 新增调度任务消费者MQ队列
  17.  */
  18. @Slf4j
  19. public class LogisticsAddDispatchMqComsumer implements MessageListenerConcurrently {
  20.  
  21.     @Override
  22.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
  23.         if (CollectionUtils.isEmpty(msgs)) {
  24.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  25.         }
  26.  
  27.         MessageExt message = msgs.get(0);
  28.         try {
  29.             String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
  30.             AddDispatchMqMessage addDispatchMqMessage = JsonUtil.toObject(messageBody, AddDispatchMqMessage.class);
  31.  
  32.             System.out.println("messageId: " + message.getMsgId() + ",topic: " +
  33.                     message.getTopic() + ",addDispatchMqMessage: " + addDispatchMqMessage);
  34.  
  35.             System.out.println(1 / 0);
  36.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  37.         } catch (Exception e) {
  38.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  39.         }
  40.     }
  41. }

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS，mq的偏移量才会下移。也就是手动ack,也只有手动返回CONSUME_SUCCESS，消息体才会偏移。

return ConsumeConcurrentlyStatus.RECONSUME_LATER;

返回 ConsumeConcurrentlyStatus.RECONSUME_LATER; mq会默认重试16次，每次执行间隔

不等。最长好像是2个多小时，具体多少自己看官方文档，一般线上环境设置重试五次失败就进入死信队列了，我这里设置的是重试三次

  1. onsumer Started.
  2. date=Fri Aug 05 14:08:52 CST 2022 *******
  3. msg=0A28A4923EC018B4AAC217A272330000
  4. date=Fri Aug 05 14:08:52 CST 2022
  5. ReconsumeTimes=0                        '第一次处理'
  6.  
  7. date=Fri Aug 05 14:09:02 CST 2022 *******
  8. msg=0A28A4923EC018B4AAC217A272330000
  9. date=Fri Aug 05 14:09:02 CST 2022
  10. ReconsumeTimes=1                       '第2次处理 与第一次间隔10s'
  11.  
  12. date=Fri Aug 05 14:09:33 CST 2022 *******
  13. msg=0A28A4923EC018B4AAC217A272330000
  14. date=Fri Aug 05 14:09:33 CST 2022
  15. ReconsumeTimes=2						'第3次处理 与第2次间隔20s'
  16.  
  17. date=Fri Aug 05 14:10:33 CST 2022 *******
  18. msg=0A28A4923EC018B4AAC217A272330000
  19. date=Fri Aug 05 14:10:33 CST 2022
  20. ReconsumeTimes=3                       '第4次处理 与第3次间隔1m'

2.自动ack封装

BaseMqMessageListener封装

  1. package com.zhjt.rocketmq.listener;
  2.  
  3.  
  4. import com.zhjt.rocketmq.constant.RocketMqSysConstant;
  5. import com.zhjt.rocketmq.domain.BaseMqMessage;
  6. import com.zhjt.rocketmq.template.RocketMqTemplate;
  7. import com.zhjt.rocketmq.utils.JsonUtil;
  8. import org.apache.rocketmq.client.producer.SendResult;
  9. import org.apache.rocketmq.client.producer.SendStatus;
  10. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  11. import org.slf4j.Logger;
  12. import org.slf4j.LoggerFactory;
  13. import org.slf4j.MDC;
  14.  
  15. import javax.annotation.Resource;
  16. import java.time.Instant;
  17. import java.util.Objects;
  18.  
  19. /**
  20.  * 抽象消息监听器，封装了所有公共处理业务，如
  21.  * 1、基础日志记录
  22.  * 2、异常处理
  23.  * 3、消息重试
  24.  * 4、警告通知
  25.  * 5、....
  26.  *
  27.  * @author 吴顺杰
  28.  * @since 2024/6/17
  29.  */
  30. public abstract class BaseMqMessageListener<T extends BaseMqMessage> {
  31.     /**
  32.      * 这里的日志记录器是哪个子类的就会被哪个子类的类进行初始化
  33.      */
  34.     protected final Logger logger = LoggerFactory.getLogger(this.getClass());
  35.     @Resource
  36.     private RocketMqTemplate rocketMqTemplate;
  37.  
  38.     /**
  39.      * 消息者名称
  40.      *
  41.      * @return 消费者名称
  42.      */
  43.     protected abstract String consumerName();
  44.  
  45.     /**
  46.      * 消息处理
  47.      *
  48.      * @param message 待处理消息
  49.      * @throws Exception 消费异常
  50.      */
  51.     protected abstract void handleMessage(T message) throws Exception;
  52.  
  53.     /**
  54.      * 超过重试次数消息，需要启用isRetry
  55.      *
  56.      * @param message 待处理消息
  57.      */
  58.     protected abstract void overMaxRetryTimesMessage(T message);
  59.     /**
  60.      * 是否过滤消息，例如某些
  61.      *
  62.      * @param message 待处理消息
  63.      * @return true: 本次消息被过滤，false：不过滤
  64.      */
  65.     protected boolean isFilter(T message) {
  66.         return false;
  67.     }
  68.  
  69.     /**
  70.      * 是否异常时重复发送
  71.      *
  72.      * @return true: 消息重试，false：不重试
  73.      */
  74.     protected abstract boolean isRetry();
  75.  
  76.     /**
  77.      * 消费异常时是否抛出异常
  78.      *
  79.      * @return true: 抛出异常，false：消费异常(如果没有开启重试则消息会被自动ack)
  80.      */
  81.     protected abstract boolean isThrowException();
  82.  
  83.     /**
  84.      * 最大重试此处
  85.      *
  86.      * @return 最大重试次数，默认10次
  87.      */
  88.     protected int maxRetryTimes() {
  89.         return 10;
  90.     }
  91.  
  92.     /**
  93.      * isRetry开启时，重新入队延迟时间
  94.      *
  95.      * @return -1：立即入队重试
  96.      */
  97.     protected int retryDelayLevel() {
  98.         return -1;
  99.     }
  100.  
  101.     /**
  102.      * 由父类来完成基础的日志和调配，下面的只是提供一个思路
  103.      */
  104.     public void dispatchMessage(T message) {
  105.         MDC.put(RocketMqSysConstant.TRACE_ID, message.getTraceId());
  106.         // 基础日志记录被父类处理了
  107.         logger.info("[{}]消费者收到消息[{}]", consumerName(), JsonUtil.toJson(message));
  108.         if (isFilter(message)) {
  109.             logger.info("消息不满足消费条件，已过滤");
  110.             return;
  111.         }
  112.         // 超过最大重试次数时调用子类方法处理
  113.         if (message.getRetryTimes() > maxRetryTimes()) {
  114.             overMaxRetryTimesMessage(message);
  115.             return;
  116.         }
  117.         try {
  118.             long start = Instant.now().toEpochMilli();
  119.             handleMessage(message);
  120.             long end = Instant.now().toEpochMilli();
  121.             logger.info("消息消费成功，耗时[{}ms]", (end - start));
  122.         } catch (Exception e) {
  123.             logger.error("消息消费异常", e);
  124.             // 是捕获异常还是抛出，由子类决定
  125.             if (isThrowException()) {
  126.                 throw new RuntimeException(e);
  127.             }
  128.             if (isRetry()) {
  129.                 // 获取子类RocketMQMessageListener注解拿到topic和tag
  130.                 RocketMQMessageListener annotation = this.getClass().getAnnotation(RocketMQMessageListener.class);
  131.                 if (Objects.nonNull(annotation)) {
  132.                     message.setSource(message.getSource() + "消息重试");
  133.                     message.setRetryTimes(message.getRetryTimes() + 1);
  134.                     SendResult sendResult;
  135.                     try {
  136.                         // 如果消息发送不成功，则再次重新发送，如果发送异常则抛出由MQ再次处理(异常时不走延迟消息)
  137.                         // 此处捕获之后，相当于此条消息被消息完成然后重新发送新的消息
  138.                         sendResult = rocketMqTemplate.send(annotation.topic(), annotation.selectorExpression(), message, retryDelayLevel());
  139.                     } catch (Exception ex) {
  140.                         throw new RuntimeException(ex);
  141.                     }
  142.                     // 发送失败的处理就是不进行ACK，由RocketMQ重试
  143.                     if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
  144.                         throw new RuntimeException("重试消息发送失败");
  145.                     }
  146.                 }
  147.             }
  148.         }
  149.     }
  150. }

IsolationConfigNew文件里面的方法：

  1. package com.zhjt.rocketmq.config;
  2.  
  3. /**
  4.  * @author: 吴顺杰
  5.  * @create: 2024-06-18 10:01
  6.  * @Description:
  7.  */
  8.  
  9. import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
  10. import org.springframework.beans.BeansException;
  11. import org.springframework.beans.factory.annotation.Value;
  12. import org.springframework.beans.factory.config.BeanPostProcessor;
  13. import org.springframework.context.annotation.Configuration;
  14. import org.springframework.lang.NonNull;
  15. import org.springframework.util.StringUtils;
  16.  
  17. /**
  18.  * RocketMQ多环境隔离配置
  19.  * 原理：对于每个配置的Bean在实例化前，拿到Bean的监听器注解把group或者topic改掉
  20.  *
  21.  * @author tianxincoord@163.com
  22.  * @since 2022/5/18
  23.  */
  24. @Configuration
  25. public class IsolationConfigNew implements BeanPostProcessor {
  26.     @Value("${system.environment.isolation:true}")
  27.     private boolean enabledIsolation;
  28.     @Value("${system.environment.name:''}")
  29.     private String environmentName;
  30.  
  31.     @Override
  32.     public Object postProcessBeforeInitialization(@NonNull Object bean,
  33.                                                   @NonNull String beanName) throws BeansException {
  34.         // DefaultRocketMQListenerContainer是监听器实现类
  35.         if (bean instanceof DefaultRocketMQListenerContainer) {
  36.             DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) bean;
  37.             // 开启消息隔离情况下获取隔离配置，隔离topic，根据自己的需求隔离group或者tag
  38.             if (enabledIsolation && StringUtils.hasText(environmentName)) {
  39.                 container.setTopic(String.join("_", container.getTopic(), environmentName));
  40.                 container.setConsumerGroup(String.join("_", container.getConsumerGroup(), environmentName));
  41.             }
  42.             return container;
  43.         }
  44.         return bean;
  45.     }
  46. }

消费者监听实现LogisticsAddDispatchMqComsumer2

  1. package com.logistics.business.rocketMq.comsumer;
  2.  
  3. import com.logistics.common.rocketMq.constant.AddDispatchMqContant;
  4. import com.logistics.common.rocketMq.domain.AddDispatchMqMessage;
  5. import com.logistics.common.rocketMq.listener.BaseMqMessageListener;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  8. import org.apache.rocketmq.spring.core.RocketMQListener;
  9. import org.springframework.stereotype.Component;
  10.  
  11. /**
  12.  * 新增调度任务消费者MQ队列
  13.  */
  14. @Slf4j
  15. @Component
  16. @RocketMQMessageListener(
  17.         topic = AddDispatchMqContant.ADD_DISPATCH_TOPIC,
  18.         consumerGroup = AddDispatchMqContant.ADD_DISPATCH_GROUP,
  19.         selectorExpression = AddDispatchMqContant.ADD_DISPATCH_GROUP,
  20.         // 指定消费者线程数，默认64，生产中请注意配置，避免过大或者过小
  21.         consumeThreadMax = 21
  22. )
  23. public class LogisticsAddDispatchMqComsumer2 extends BaseMqMessageListener<AddDispatchMqMessage>
  24.         implements RocketMQListener<AddDispatchMqMessage> {
  25.  
  26.     /**
  27.      * 此处只是说明封装的思想，更多还是要根据业务操作决定
  28.      * 内功心法有了，无论什么招式都可以发挥最大威力
  29.      */
  30.     @Override
  31.     protected String consumerName() {
  32.         return "RocketMq监听消息";
  33.     }
  34.  
  35.     @Override
  36.     public void onMessage(AddDispatchMqMessage message) {
  37.         // 注意，此时这里没有直接处理业务，而是先委派给父类做基础操作，然后父类做完基础操作后会调用子类的实际处理类型
  38.         super.dispatchMessage(message);
  39.     }
  40.  
  41.     @Override
  42.     protected void handleMessage(AddDispatchMqMessage message) throws Exception {
  43.         // 此时这里才是最终的业务处理，代码只需要处理资源类关闭异常，其他的可以交给父类重试
  44.         // 业务异常直接抛出异常即可 否则捕获异常没有抛出 无法进行重试
  45.         log.info("RocketMq监消息消费数据：{}", message);
  46.     }
  47.  
  48.     @Override
  49.     protected void overMaxRetryTimesMessage(AddDispatchMqMessage message) {
  50.         // 当超过指定重试次数消息时此处方法会被调用
  51.         // 生产中可以进行回退或其他业务操作
  52.     }
  53.  
  54.     @Override
  55.     protected boolean isRetry() {
  56.         return true;
  57.     }
  58.  
  59.     @Override
  60.     protected int maxRetryTimes() {
  61.         // 指定需要的重试次数，超过重试次数overMaxRetryTimesMessage会被调用
  62.         return 5;
  63.     }
  64.  
  65.     @Override
  66.     protected boolean isThrowException() {
  67.         // 是否抛出异常，到消费异常时是被父类拦截处理还是直接抛出异常
  68.         return false;
  69.     }
  70. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/运维做开发/article/detail/988457
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号