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

Spring Boot 通过监听器方式整合 RocketMq（基于模板方法的并发消费、局部顺序消息消费）_springboot rocketmq 消费者监听

作者：AllinToyou | 2024-06-13 13:43:24

赞

踩

springboot rocketmq 消费者监听

目录

生产者发送消息

消费者消费消息

常见问题

 1、消息发送失败处理方式

2、消费过程幂等

3、消费速度慢的处理方式

 Broker 角色

项目中有关rocketMq及相关类的Maven依赖

  1.         <dependency>
  2.             <groupId>org.apache.rocketmq</groupId>
  3.             <artifactId>rocketmq-spring-boot-starter</artifactId>
  4.             <version>2.1.1</version>
  5.         </dependency>
  6.         <dependency>
  7.             <groupId>org.apache.rocketmq</groupId>
  8.             <artifactId>rocketmq-client</artifactId>
  9.             <version>4.3.0</version>
  10.         </dependency>
  11.         <dependency>
  12.             <groupId>com.google.guava</groupId>
  13.                 <artifactId>guava</artifactId>
  14.             <version>30.1.1-jre</version>
  15.         </dependency>

生产者发送消息

并发消费场景下的生产者代码，并发消息无法保证消息一定是按照顺序消费，在绝大多数场景下不需要过问消息的消费顺序，可通过此方式进行mq消息的发送：

  1. package com.fss.project.mq.producer;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.fss.project.mq.enums.DelayLevelEnum;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.apache.rocketmq.client.exception.MQBrokerException;
  7. import org.apache.rocketmq.client.exception.MQClientException;
  8. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  9. import org.apache.rocketmq.client.producer.SendResult;
  10. import org.apache.rocketmq.client.producer.SendStatus;
  11. import org.apache.rocketmq.common.message.Message;
  12. import org.apache.rocketmq.remoting.exception.RemotingException;
  13. import org.slf4j.Logger;
  14. import org.slf4j.LoggerFactory;
  15. import org.springframework.stereotype.Component;
  16.  
  17. import javax.annotation.Resource;
  18. import java.nio.charset.StandardCharsets;
  19. import java.util.Objects;
  20.  
  21. @Slf4j
  22. @Component
  23. public class RocketMqProductComponent {
  24.  
  25.     private Logger logger = LoggerFactory.getLogger(RocketMqProductComponent.class);
  26.  
  27.     @Resource
  28.     private DefaultMQProducer defaultMQProducer;
  29.  
  30.     /**
  31.      * 发送消息
  32.      *
  33.      * @param dto       具体数据
  34.      * @param topicName
  35.      * @param tagName
  36.      * @param key
  37.      * @return 执行状态
  38.      */
  39.     public boolean sendMessage(Object dto, String topicName, String tagName, String key) {
  40.         if (Objects.isNull(dto) || Objects.isNull(topicName) || Objects.isNull(tagName)) {
  41.             return false;
  42.         }
  43.  
  44.         boolean result = false;
  45.         // 构造消息body
  46.         String body = builderBody(dto);
  47.  
  48.         try {
  49.             Message message = new Message(topicName, tagName, key, body.getBytes(StandardCharsets.UTF_8));
  50.             SendResult send = defaultMQProducer.send(message);
  51.  
  52.             logger.info("发送者，发送消息：" + JSON.toJSONString(send));
  53.             if (Objects.nonNull(send) && SendStatus.SEND_OK.equals(send.getSendStatus())) {
  54.                 result = true;
  55.             } else {
  56.                 logger.warn("消息发送失败,send={},body={}", JSON.toJSONString(send), body);
  57.             }
  58.         } catch (MQClientException | RemotingException | MQBrokerException e) {
  59.             logger.warn("发送消息失败：｛｝", e);
  60.         } catch (InterruptedException e) {
  61.             Thread.currentThread().interrupt();
  62.         }
  63.         return result;
  64.     }
  65.  
  66.     /**
  67.      * 发送延时消息
  68.      *
  69.      * @param dto
  70.      * @param topicName
  71.      * @param tagName
  72.      * @param delayLevelEnum 延时等级
  73.      * @return
  74.      */
  75.     public boolean sendDelayMessage(Object dto, String topicName, String tagName, String key, DelayLevelEnum delayLevelEnum) {
  76.         if (Objects.isNull(dto) || Objects.isNull(topicName) || Objects.isNull(tagName) || Objects.isNull(delayLevelEnum)) {
  77.             return false;
  78.         }
  79.  
  80.         boolean result = false;
  81.         // 构造消息body
  82.         String body = builderBody(dto);
  83.  
  84.         try {
  85.             Message message = new Message(topicName, tagName, key, body.getBytes(StandardCharsets.UTF_8));
  86.             message.setDelayTimeLevel(delayLevelEnum.getDelayLevel());
  87.             logger.warn("发送延时消息 message：｛｝", JSON.toJSONString(message));
  88.             SendResult send = defaultMQProducer.send(message);
  89.             if (Objects.nonNull(send) && SendStatus.SEND_OK.equals(send.getSendStatus())) {
  90.                 result = true;
  91.             } else {
  92.                 logger.warn("延时消息发送失败,send={},body={}", JSON.toJSONString(send), body);
  93.             }
  94.         } catch (MQClientException | RemotingException | MQBrokerException e) {
  95.             logger.warn("发送延时消息失败：｛｝", e);
  96.         } catch (InterruptedException e) {
  97.             Thread.currentThread().interrupt();
  98.         }
  99.         return result;
  100.     }
  101.  
  102.     /**
  103.      * 构造消息body
  104.      *
  105.      * @param dto
  106.      * @return
  107.      */
  108.     public String builderBody(Object dto) {
  109.         // 构造消息body
  110.         String body = null;
  111.         if (dto instanceof String) {
  112.             body = (String) dto;
  113.         } else {
  114.             body = JSON.toJSONString(dto);
  115.         }
  116.         return body;
  117.     }
  118. }

顺序消费场景下的生产者代码（局部顺序），在特定场景下需要消息按照顺序消费时，可通过此方式进行发送

  1. package com.fss.project.mq.producer;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.fss.project.mq.enums.DelayLevelEnum;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.apache.rocketmq.client.exception.MQBrokerException;
  7. import org.apache.rocketmq.client.exception.MQClientException;
  8. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  9. import org.apache.rocketmq.client.producer.MessageQueueSelector;
  10. import org.apache.rocketmq.client.producer.SendResult;
  11. import org.apache.rocketmq.client.producer.SendStatus;
  12. import org.apache.rocketmq.common.message.Message;
  13. import org.apache.rocketmq.common.message.MessageQueue;
  14. import org.apache.rocketmq.remoting.exception.RemotingException;
  15. import org.slf4j.Logger;
  16. import org.slf4j.LoggerFactory;
  17. import org.springframework.stereotype.Component;
  18.  
  19. import javax.annotation.Resource;
  20. import java.nio.charset.StandardCharsets;
  21. import java.util.List;
  22. import java.util.Objects;
  23.  
  24. @Slf4j
  25. @Component
  26. public class RocketMqProductOrderLyComponent {
  27.  
  28.     private Logger logger = LoggerFactory.getLogger(RocketMqProductOrderLyComponent.class);
  29.  
  30.     @Resource
  31.     private DefaultMQProducer defaultMQProducer;
  32.  
  33.     /**
  34.      * 发送消息
  35.      *
  36.      * @param dto       具体数据
  37.      * @param topicName
  38.      * @param tagName
  39.      * @return 执行状态
  40.      */
  41.     public boolean sendMessage(Object dto, String topicName, String tagName, String key) {
  42.         if (Objects.isNull(dto) || Objects.isNull(topicName) || Objects.isNull(tagName) || Objects.isNull(key)) {
  43.             return false;
  44.         }
  45.  
  46.         boolean result = false;
  47.  
  48.         // 构造消息body
  49.         String body = builderBody(dto);
  50.  
  51.         try {
  52.             Message message = new Message(topicName, tagName, key, body.getBytes(StandardCharsets.UTF_8));
  53.             /**
  54.              * 局部的顺序消息
  55.              * message:消息信息
  56.              * arg：选择队列的业务标识
  57.              */
  58.             SendResult send = defaultMQProducer.send(message, new MessageQueueSelector() {
  59.                 @Override
  60.                 public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
  61.                     Long key = Long.parseLong((String) o);
  62.                     int index = (int) (key % list.size());
  63.                     return list.get(index);
  64.                 }
  65.             }, key);
  66.  
  67.             System.err.println("发送者，发送消息：" + JSON.toJSONString(send));
  68.             if (Objects.nonNull(send) && SendStatus.SEND_OK.equals(send.getSendStatus())) {
  69.                 result = true;
  70.             } else {
  71.                 logger.warn("消息发送失败,send={},body={}", JSON.toJSONString(send), body);
  72.             }
  73.         } catch (MQClientException | RemotingException | MQBrokerException e) {
  74.             logger.warn("发送消息失败：｛｝", e);
  75.         } catch (InterruptedException e) {
  76.             Thread.currentThread().interrupt();
  77.         }
  78.         return result;
  79.     }
  80.  
  81.     /**
  82.      * 发送延时消息
  83.      *
  84.      * @param dto
  85.      * @param topicName
  86.      * @param tagName
  87.      * @param delayLevelEnum 延时等级
  88.      * @return
  89.      */
  90.     public boolean sendDelayMessage(Object dto, String topicName, String tagName, String key, DelayLevelEnum delayLevelEnum) {
  91.         if (Objects.isNull(dto) || Objects.isNull(topicName) || Objects.isNull(tagName) || Objects.isNull(key) || Objects.isNull(delayLevelEnum)) {
  92.             return false;
  93.         }
  94.  
  95.         boolean result = false;
  96.  
  97.         // 构造消息body
  98.         String body = builderBody(dto);
  99.         try {
  100.             Message message = new Message(topicName, tagName, key, body.getBytes(StandardCharsets.UTF_8));
  101.             message.setDelayTimeLevel(delayLevelEnum.getDelayLevel());
  102.  
  103.             SendResult send = defaultMQProducer.send(message, new MessageQueueSelector() {
  104.                 @Override
  105.                 public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
  106.                     Long key = (Long) o;
  107.                     int index = (int) (key % list.size());
  108.                     return list.get(index);
  109.                 }
  110.             }, key);
  111.             if (Objects.nonNull(send) && SendStatus.SEND_OK.equals(send.getSendStatus())) {
  112.                 result = true;
  113.             } else {
  114.                 logger.warn("延时消息发送失败,send={},body={}", JSON.toJSONString(send), body);
  115.             }
  116.         } catch (MQClientException | RemotingException | MQBrokerException e) {
  117.             logger.warn("发送延时消息失败：｛｝", e);
  118.         } catch (InterruptedException e) {
  119.             Thread.currentThread().interrupt();
  120.         }
  121.         return result;
  122.     }
  123.  
  124.     /**
  125.      * 构造消息body
  126.      *
  127.      * @param dto
  128.      * @return
  129.      */
  130.     public String builderBody(Object dto) {
  131.         // 构造消息body
  132.         String body = null;
  133.         if (dto instanceof String) {
  134.             body = (String) dto;
  135.         } else {
  136.             body = JSON.toJSONString(dto);
  137.         }
  138.         return body;
  139.     }
  140. }

消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为分区有序或者全局有序。

顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。

通常局部有序已经我完全可以满足需求，并且效率上会更高，故通常是使用局部消费

所以在代码中，要发送顺序消息时，须指定key，即作为message的key，也作为key对队列长度取余来选择某个queue，从而实现局部顺序，key可以传入业务的唯一ID，例：订单ID、退款ID等。

消费者消费消息

消费者是通过模板方法的方式，来让开发者更多的关注业务逻辑，在监听器中对消息已经有了统一的处理。

消费者Bean配置

  1. package com.fss.project.mq.consume;
  2.  
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  6. import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
  7. import org.slf4j.Logger;
  8. import org.slf4j.LoggerFactory;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.beans.factory.annotation.Value;
  11. import org.springframework.boot.SpringBootConfiguration;
  12. import org.springframework.context.annotation.Bean;
  13. import org.springframework.util.StringUtils;
  14. import java.util.List;
  15.  
  16. /**
  17.  * 消费者配置
  18.  */
  19. @SpringBootConfiguration
  20. public class MQConsumerConfiguration {
  21.     public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumerConfiguration.class);
  22.  
  23.  
  24.     @Value("${rocketmq.consumer.namesrvAddr:127.0.0.1:9876}")
  25.     private String namesrvAddr;
  26.  
  27.     @Value("${rocketmq.consumer.groupName:producer-group}")
  28.     private String groupName;
  29.  
  30.     /**
  31.      * 并发消费topic
  32.      */
  33.     @Value("#{'${rocketmq.consumer.topics:DEMO_TEST_TOPIC}'.split(',')}")
  34.     private List<String> topicList;
  35.  
  36.     /**
  37.      * 顺序消费topic
  38.      */
  39.     @Value("#{'${rocketmq.consumer.topics:DEMO_ORDERLY_TOPIC}'.split(',')}")
  40.     private List<String> orderLyTopicList;
  41.  
  42.     @Value("${rocketmq.consumer.consumeMessageBatchMaxSize:1}")
  43.     private int consumeMessageBatchMaxSize;
  44.  
  45.     /**
  46.      * 并发消费监听器
  47.      */
  48.     @Autowired
  49.     private RocketMqMessageListener registerMessageListener;
  50.  
  51.     /**
  52.      * 顺序消费监听器
  53.      */
  54.     @Autowired
  55.     private RocketMqMessageOrderLyListener rocketMqMessageOrderLyListener;
  56.  
  57.     @Bean
  58.     public DefaultMQPushConsumer getRocketMQConsumer() throws RuntimeException {
  59.         if (StringUtils.isEmpty(groupName)){
  60.             throw new RuntimeException("groupName is null !!!");
  61.         }
  62.         if (StringUtils.isEmpty(namesrvAddr)){
  63.             throw new RuntimeException("namesrvAddr is null !!!");
  64.         }
  65.         if(StringUtils.isEmpty(topicList)){
  66.             throw new RuntimeException("topics is null !!!");
  67.         }
  68.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
  69.         consumer.setNamesrvAddr(namesrvAddr);
  70.         consumer.registerMessageListener(registerMessageListener);
  71. //        consumer.registerMessageListener(rocketMqMessageOrderLyListener);
  72.         /**
  73.          * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
  74.          * 如果非第一次启动，那么按照上次消费的位置继续消费
  75.          */
  76.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  77.         /**
  78.          * 设置消费模型，集群还是广播，默认为集群
  79.          */
  80.         consumer.setMessageModel(MessageModel.CLUSTERING);
  81.         /**
  82.          * 设置一次消费消息的条数，默认为1条
  83.          */
  84.         consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
  85.         try {
  86.             /**
  87.              * 设置该消费者订阅的主题和tag，如果是订阅该主题下的所有tag，则tag使用*；如果需要指定订阅该主题下的某些tag，则使用||分割，例如tag1||tag2||tag3
  88.              */
  89.             topicList.forEach(topic->{
  90.                 try {
  91.                     consumer.subscribe(topic,"*");
  92.                 } catch (MQClientException e) {
  93.                     e.printStackTrace();
  94.                 }
  95.             });
  96. //            orderLyTopicList.forEach(topic->{
  97. //                try {
  98. //                    consumer.subscribe(topic,"*");
  99. //                } catch (MQClientException e) {
  100. //                    e.printStackTrace();
  101. //                }
  102. //            });
  103.             consumer.start();
  104.             LOGGER.info("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topicList,namesrvAddr);
  105.         }catch (MQClientException e){
  106.             LOGGER.error("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topicList,namesrvAddr,e);
  107.             throw new RuntimeException(e);
  108.         }
  109.         return consumer;
  110.     }
  111.  
  112. }

通过DefaultMQPushConsumer设置监听器的实现类，来将消费逻辑转移给监听器RocketMqMessageListener 。

MessageHandler:定义处理消息的接口handle及处理的消息topic、tag的类型。

  1. public interface MessageHandler {
  2.  
  3.     void handle(String body);
  4.  
  5.     List<String> tags();
  6.  
  7.     String topic();
  8. }
AbstractMessageHandler：MessageHandler的子类，主要是自动将body的数据转成对应的dto

  1. package com.fss.project.mq.consume;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4.  
  5. import java.lang.reflect.ParameterizedType;
  6. import java.lang.reflect.Type;
  7.  
  8. public abstract class AbstractMessageHandler<T> implements MessageHandler{
  9.  
  10.     private Class<T> paramClass;
  11.  
  12.     public AbstractMessageHandler() {
  13.         Type genericSuperclass = this.getClass().getGenericSuperclass();
  14.         if(genericSuperclass instanceof ParameterizedType){
  15.             ParameterizedType parameterizedType = (ParameterizedType) genericSuperclass;
  16.             Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
  17.             paramClass = (Class<T>) actualTypeArguments[0];
  18.         }
  19.     }
  20.  
  21.     @Override
  22.     public void handle(String body) {
  23.         T param = JSON.parseObject(body, paramClass);
  24.         if (checkDo(param)) {
  25.             handler(param);
  26.         }
  27.     }
  28.  
  29.     /**
  30.      * 执行消费的具体业务逻辑
  31.      * @param param
  32.      */
  33.     public abstract void handler(T param);
  34.  
  35.     /**
  36.      * 检查是否需要执行
  37.      */
  38.     boolean checkDo(T param){
  39.         return true;
  40.     }
  41. }

RocketMqMessageListener ：通过topic、tag可以确定AbstractMessageHandler的子类（某个topic、tag消费者），若传入的有key，则可以根据key来防止重复消费消息（key通常作为唯一业务），如没有传入，在具体的消费业务代码中根据业务ID来特定处理也可以。

  1. package com.fss.project.mq.consume;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.google.common.collect.HashBasedTable;
  5. import com.google.common.collect.Table;
  6. import org.apache.commons.collections.CollectionUtils;
  7. import org.apache.commons.lang3.StringUtils;
  8. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  9. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  10. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  11. import org.apache.rocketmq.common.message.MessageExt;
  12. import org.slf4j.Logger;
  13. import org.slf4j.LoggerFactory;
  14. import org.springframework.beans.BeansException;
  15. import org.springframework.beans.factory.annotation.Value;
  16. import org.springframework.context.ApplicationContext;
  17. import org.springframework.context.ApplicationContextAware;
  18. import org.springframework.stereotype.Component;
  19.  
  20. import javax.annotation.PostConstruct;
  21. import java.nio.charset.StandardCharsets;
  22. import java.util.*;
  23.  
  24. @Component
  25. public class RocketMqMessageListener implements MessageListenerConcurrently, ApplicationContextAware {
  26.  
  27.     private Logger logger = LoggerFactory.getLogger(RocketMqMessageListener.class);
  28.  
  29.     private ApplicationContext context;
  30.  
  31.     @Value("#{'${rocketmq.consumer.topics:DEMO_TEST_TOPIC}'.split(',')}")
  32.     private List<String> topicList;
  33.  
  34.     private Table<String, String, List<MessageHandler>> messageHandlerTable = HashBasedTable.create();
  35.  
  36.     @PostConstruct
  37.     public void init() {
  38.         Map<String, MessageHandler> consumers = context.getBeansOfType(MessageHandler.class);
  39.         consumers.values().forEach(
  40.                 messageHandler -> {
  41.                     String topic = messageHandler.topic();
  42.                     for (String tagName : messageHandler.tags()) {
  43.                         List<MessageHandler> messageHandlers = messageHandlerTable.get(topic, tagName);
  44.                         if (messageHandlers == null) {
  45.                             messageHandlers = new ArrayList<>(3);
  46.                         }
  47.                         messageHandlers.add(messageHandler);
  48.                         messageHandlerTable.put(topic, tagName, messageHandlers);
  49.                     }
  50.                 }
  51.         );
  52.     }
  53.  
  54.     private List<MessageHandler> getHandler(String topic, String tag) {
  55.         if (StringUtils.isBlank(topic) || StringUtils.isBlank(tag)) {
  56.             return Collections.emptyList();
  57.         }
  58.  
  59.         return messageHandlerTable.get(topic, tag);
  60.     }
  61.  
  62.     @Override
  63.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
  64.         try {
  65.             for (MessageExt messageExt : list) {
  66.                 handlerMessageExt(messageExt);
  67.                 logger.info("ThreadName:{},messageExt:{}，消费成功", Thread.currentThread().getName(), messageExt);
  68.             }
  69.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  70.         } catch (Exception e) {
  71.             logger.warn("消息消费异常：e:{}", e);
  72.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  73.         }
  74.     }
  75.  
  76.     private void handlerMessageExt(MessageExt messageExt) {
  77.         String topic = messageExt.getTopic();
  78.         String tag = messageExt.getTags();
  79.         String key = messageExt.getKeys();
  80.  
  81.         // 若传入key，则做唯一性校验
  82.         if (Objects.nonNull(key) && checkMessageKey(key)) {
  83.  
  84.         }
  85.  
  86.         String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
  87.         logger.info("handlerMessageExt,topic:{},tag:{},body:{}", topic, tag, body);
  88.         if (!topicList.contains(topic)) {
  89.             return;
  90.         }
  91.  
  92.         List<MessageHandler> messageHandlerList = getHandler(topic, tag);
  93.         if (CollectionUtils.isNotEmpty(messageHandlerList)) {
  94.             messageHandlerList.forEach(
  95.                     messageHandler -> {
  96.                         messageHandler.handle(body);
  97.                     }
  98.             );
  99.         }
  100.  
  101.     }
  102.  
  103.     /**
  104.      * 判断是否重复消费
  105.      */
  106.     private boolean checkMessageKey(String key) {
  107.         // TODO: 2021/9/4 可根据redis、数据库保证消息不重复消费
  108.         return false;
  109.     }
  110.  
  111.     @Override
  112.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  113.         context = applicationContext;
  114.     }
  115. }

消费业务demo

  1. package com.fss.project.mq.consume.component;
  2.  
  3. import com.fss.project.mq.consume.AbstractMessageHandler;
  4. import com.fss.project.mq.consume.RocketMqTagEnum;
  5. import com.fss.project.mq.consume.TopicEnum;
  6. import com.fss.project.mq.dto.OrderDto;
  7. import org.springframework.stereotype.Component;
  8. import java.util.Collections;
  9. import java.util.List;
  10.  
  11. @Component
  12. public class OrderConsumer extends AbstractMessageHandler<OrderDto> {
  13.  
  14.     @Override
  15.     public void handler(OrderDto param) {
  16.         if(checkConsumer()){
  17.             return;
  18.         }
  19.         System.err.println("成功消费："+param);
  20.     }
  21.  
  22.     /**
  23.      * 若message未设置key，则可以根据body的数据（例orderId、refundId），根据redis、数据库中当前数据的状态来判断是否要消费
  24.      * 若数据没有唯一ID来区分，则可以认为该消息不重要不作为唯一校验
  25.      * @return
  26.      */
  27.     private boolean checkConsumer() {
  28.         return false;
  29.     }
  30.  
  31.     @Override
  32.     public List<String> tags() {
  33.         return Collections.singletonList(RocketMqTagEnum.TEST_TAG.getTagName());
  34.     }
  35.  
  36.     @Override
  37.     public String topic() {
  38.         return TopicEnum.TEST_TOPIC.getTopicName();
  39.     }
  40. }

若要顺序消费，消费者监听类需要实现MessageListenerOrderly来实现，例：RocketMqMessageOrderLyListener

  1. package com.fss.project.mq.consume;
  2.  
  3. import com.google.common.collect.HashBasedTable;
  4. import com.google.common.collect.Table;
  5. import org.apache.commons.collections.CollectionUtils;
  6. import org.apache.commons.lang3.StringUtils;
  7. import org.apache.rocketmq.client.consumer.listener.*;
  8. import org.apache.rocketmq.common.message.MessageExt;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11. import org.springframework.beans.BeansException;
  12. import org.springframework.beans.factory.annotation.Value;
  13. import org.springframework.context.ApplicationContext;
  14. import org.springframework.context.ApplicationContextAware;
  15. import org.springframework.stereotype.Component;
  16.  
  17. import javax.annotation.PostConstruct;
  18. import java.nio.charset.StandardCharsets;
  19. import java.util.*;
  20.  
  21. @Component
  22. public class RocketMqMessageOrderLyListener implements MessageListenerOrderly, ApplicationContextAware {
  23.  
  24.     private Logger logger = LoggerFactory.getLogger(RocketMqMessageOrderLyListener.class);
  25.  
  26.     private ApplicationContext context;
  27.  
  28.     @Value("#{'${rocketmq.consumer.orderly.topics:DEMO_ORDERLY_TOPIC}'.split(',')}")
  29.     private List<String> topicList;
  30.  
  31.     private Table<String, String, List<MessageHandler>> messageHandlerTable = HashBasedTable.create();
  32.  
  33.     @PostConstruct
  34.     public void init() {
  35.         Map<String, MessageHandler> consumers = context.getBeansOfType(MessageHandler.class);
  36.         consumers.values().forEach(
  37.                 messageHandler -> {
  38.                     String topic = messageHandler.topic();
  39.                     for (String tagName : messageHandler.tags()) {
  40.                         List<MessageHandler> messageHandlers = messageHandlerTable.get(topic, tagName);
  41.                         if (messageHandlers == null) {
  42.                             messageHandlers = new ArrayList<>(3);
  43.                         }
  44.                         messageHandlers.add(messageHandler);
  45.                         messageHandlerTable.put(topic, tagName, messageHandlers);
  46.                     }
  47.                 }
  48.         );
  49.     }
  50.  
  51.     private List<MessageHandler> getHandler(String topic, String tag) {
  52.         if (StringUtils.isBlank(topic) || StringUtils.isBlank(tag)) {
  53.             return Collections.emptyList();
  54.         }
  55.  
  56.         return messageHandlerTable.get(topic, tag);
  57.     }
  58.  
  59.     private void handlerMessageExt(MessageExt messageExt) {
  60.         String topic = messageExt.getTopic();
  61.         String tag = messageExt.getTags();
  62.         String key = messageExt.getKeys();
  63.  
  64.         // 若传入key，则做唯一性校验
  65.         if (Objects.nonNull(key) && checkMessageKey(key)) {
  66.  
  67.         }
  68.  
  69.         String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
  70.         logger.info("handlerMessageExt,topic:{},tag:{},body:{}", topic, tag, body);
  71.         if (!topicList.contains(topic)) {
  72.             return;
  73.         }
  74.  
  75.         List<MessageHandler> messageHandlerList = getHandler(topic, tag);
  76.         if (CollectionUtils.isNotEmpty(messageHandlerList)) {
  77.             messageHandlerList.forEach(
  78.                     messageHandler -> {
  79.                         messageHandler.handle(body);
  80.                     }
  81.             );
  82.         }
  83.  
  84.     }
  85.  
  86.     /**
  87.      * 判断是否重复消费
  88.      */
  89.     private boolean checkMessageKey(String key) {
  90.         // TODO: 2021/9/4 可根据redis、数据库保证消息不重复消费
  91.         return false;
  92.     }
  93.  
  94.     @Override
  95.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  96.         context = applicationContext;
  97.     }
  98.  
  99.     @Override
  100.     public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
  101.         try {
  102.             for (MessageExt messageExt : list) {
  103.                 handlerMessageExt(messageExt);
  104.                 logger.info("ThreadName:{},messageExt:{}，消费成功", Thread.currentThread().getName(), messageExt);
  105.             }
  106.             return ConsumeOrderlyStatus.SUCCESS;
  107.         } catch (Exception e) {
  108.             logger.warn("消息消费异常：e:{}", e);
  109.             return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
  110.         }
  111.     }
  112. }

常见问题

 1、消息发送失败处理方式

Producer的send方法本身支持内部重试，重试逻辑如下：

  • 至多重试2次。
  • 如果同步模式发送失败，则轮转到下一个Broker，如果异步模式发送失败，则只会在当前Broker进行重试。这个方法的总耗时时间不超过sendMsgTimeout设置的值，默认10s。
  • 如果本身向broker发送消息产生超时异常，就不会再重试。

以上策略也是在一定程度上保证了消息可以发送成功。如果业务对消息可靠性要求比较高，建议应用增加相应的重试逻辑：比如调用send同步方法发送失败时，则尝试将消息存储到db，然后由后台线程定时重试，确保消息一定到达Broker。

2、消费过程幂等

RocketMQ无法避免消息重复（Exactly-Once），所以如果业务对消费重复非常敏感，务必要在业务层面进行去重处理。可以借助关系数据库进行去重。首先需要确定消息的唯一键，可以是msgId，也可以是消息内容中的唯一标识字段，例如订单Id等。在消费之前判断唯一键是否在关系数据库中存在。如果不存在则插入，并消费，否则跳过。（实际过程要考虑原子性问题，判断是否存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过）

msgId一定是全局唯一标识符，但是实际使用中，可能会存在相同的消息有两个不同msgId的情况（消费者主动重发、因客户端重投机制导致的重复等），这种情况就需要使业务字段进行重复消费。

3、消费速度慢的处理方式

1 提高消费并行度

绝大部分消息消费行为都属于 IO 密集型，即可能是操作数据库，或者调用 RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，通过增加消费并行度，可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。所以，应用必须要设置合理的并行度。 如下有几种修改消费并行度的方法：

  • 同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。可以通过加机器，或者在已有机器启动多个进程的方式。
  • 提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax实现。

2 批量方式消费

某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时 1 s，一次处理 10 个订单可能也只耗时 2 s，这样即可大幅度提高消费的吞吐量，通过设置 consumer的 consumeMessageBatchMaxSize 返个参数，默认是 1，即一次只消费一条消息，例如设置为 N，那么每次消费的消息数小于等于 N。

3 优化每条消息消费过程

举例如下，某条消息的消费过程如下：

  • 根据消息从 DB 查询【数据 1】
  • 根据消息从 DB 查询【数据 2】
  • 复杂的业务计算
  • 向 DB 插入【数据 3】
  • 向 DB 插入【数据 4】

这条消息的消费过程中有4次与 DB的 交互，如果按照每次 5ms 计算，那么总共耗时 20ms，假设业务计算耗时 5ms，那么总过耗时 25ms，所以如果能把 4 次 DB 交互优化为 2 次，那么总耗时就可以优化到 15ms，即总体性能提高了 40%。所以应用如果对时延敏感的话，可以把DB部署在SSD硬盘，相比于SCSI磁盘，前者的RT会小很多。

消费慢的两种类型：CPU内部计算代码、外部I/O操作代码

通常是外部IO操作导致的：

1、数据库操作慢

2、缓存数据库IO操作慢

3、下游系统PRC请求，响应慢

消息堆积：下游服务异常、达到DBMS的容量限制

 Broker 角色

​ Broker 角色分为 ASYNC_MASTER（异步主机）、SYNC_MASTER（同步主机）以及SLAVE（从机）。如果对消息的可靠性要求比较严格，可以采用 SYNC_MASTER加SLAVE的部署方式。如果对消息可靠性要求不高，可以采用ASYNC_MASTER加SLAVE的部署方式。如果只是测试方便，则可以选择仅ASYNC_MASTER或仅SYNC_MASTER的部署方式。

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

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

  

闽ICP备14008679号