git地址： https://github.com/leezhang0525/boot
branch：1.1.3-rocketMQ
本文档持续更新，目标为多组件系统，常用组件集成完毕后，后续会添加分布式内容。

一、说明及准备

本文测试需要依赖rocketMQ的下载以及可视化页面，下载安装流程参考https://www.cnblogs.com/sexintercourse/p/15519752.html
前置准备
1、启动rocketMQ服务
依次启动 mqnamesrv.cmd 以及 mqbroker.cmd ，cmd窗口不要关闭
2、启动rocketmq-externals
1650696034(1).png
这里可以直接maven打包，运行jar 是一样的。
访问：http://127.0.0.1:8082/#/
1650696160(1).png

二、添加依赖及修改yaml文件

  1. <dependency>
  2.   <groupId>org.apache.rocketmq</groupId>
  3.   <artifactId>rocketmq-spring-boot-starter</artifactId>
  4.   <version>${rocketmq.version}</version>
  5. </dependency>
  6. <rocketmq.version>2.0.4</rocketmq.version>
  1. server:
  2.   port: 8081
  3. spring:
  4.   datasource:
  5.     master:
  6.       url: jdbc:mysql://localhost:3306/pay?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&useLocalSessionState=true
  7.       username: root
  8.       password: 123456
  9.       driver-class-name: com.mysql.jdbc.Driver
  10.       type: com.alibaba.druid.pool.DruidDataSource
  11.     slave:
  12.       url: jdbc:mysql://localhost:3306/payslave?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&useLocalSessionState=true
  13.       username: root
  14.       password: 123456
  15.       driver-class-name: com.mysql.jdbc.Driver
  16.       type: com.alibaba.druid.pool.DruidDataSource
  17.   redis:
  18.     host: localhost
  19.     port: 6379
  20.     database: 0
  21.     password: 123456
  22.     timeout: 3S
  23.   rocket:
  24.     #生产者
  25.     producer:
  26.       namesrvAddr: localhost:9876
  27.       producerGroupName: zhangsan_producer
  28.       # 消息最大长度 默认 1024 * 4 (4M)
  29.       maxMessageSize: 4096
  30.       # 发送消息超时时间，默认 3000
  31.       sendMsgTimeOut: 3000
  32.       retryTimesWhenSendFailed: 2
  33.     #消费者
  34.     consumer:
  35.       namesrvAddr: localhost:9876
  36.       consumerGroupName: zhangsan_consumer
  37.       topicName: pay_success
  38.       tag: tag1
  39.       # 消费者线程数据量
  40.       consumeThreadMin: 5
  41.       consumeThreadMax: 32
  42.       # 设置一次消费消息的条数，默认1
  43.       consumeMessageBatchMaxSize: 1
  45. logging:
  46.   level:
  47.     root: debug

三、添加生产者，消费者

  1. package com.zhangsan.boot.mq.rocketmq;
  3. import lombok.Getter;
  4. import lombok.Setter;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  7. import org.springframework.boot.context.properties.ConfigurationProperties;
  8. import org.springframework.context.annotation.Bean;
  9. import org.springframework.context.annotation.Configuration;
  11. @Getter
  12. @Setter
  13. @Slf4j
  14. @Configuration
  15. @ConfigurationProperties(prefix = "spring.rocket.producer")
  16. public class Producer {
  18.     private String producerGroupName;
  20.     private String namesrvAddr;
  21.     // 消息最大值
  22.     private Integer maxMessageSize;
  23.     // 消息发送超时时间
  24.     private Integer sendMsgTimeOut;
  25.     // 失败重试次数
  26.     private Integer retryTimesWhenSendFailed;
  28.     /**
  29.     * 生产者
  30.     * @return
  31.     */
  32.     @Bean
  33.     public DefaultMQProducer defaultProducer() {
  34.         DefaultMQProducer producer = new DefaultMQProducer(producerGroupName);
  35.         producer.setNamesrvAddr(namesrvAddr);
  36.         producer.setVipChannelEnabled(false);
  37.         producer.setMaxMessageSize(maxMessageSize);
  38.         producer.setSendMsgTimeout(sendMsgTimeOut);
  39.         producer.setRetryTimesWhenSendAsyncFailed(retryTimesWhenSendFailed);
  40.         return producer;
  41.     }
  42. }

消费者

  1. package com.zhangsan.boot.mq.rocketmq;
  3. import com.zhangsan.boot.listener.OrderlyListener;
  4. import lombok.Getter;
  5. import lombok.Setter;
  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.common.consumer.ConsumeFromWhere;
  10. import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
  11. import org.springframework.beans.factory.annotation.Autowired;
  12. import org.springframework.boot.context.properties.ConfigurationProperties;
  13. import org.springframework.context.annotation.Bean;
  14. import org.springframework.context.annotation.Configuration;
  16. @Getter
  17. @Setter
  18. @Slf4j
  19. @Configuration
  20. @ConfigurationProperties(prefix = "spring.rocket.consumer")
  21. public class Consumer {
  23.     private String namesrvAddr;
  25.     private String consumerGroupName;
  27.     private String topicName;
  29.     private String tag;
  31.     // 消费者线程数据量
  32.     private Integer consumeThreadMin;
  33.     private Integer consumeThreadMax;
  34.     private Integer consumeMessageBatchMaxSize;
  36.     @Autowired
  37.     private OrderlyListener orderlyListener;
  39.     @Bean(initMethod = "start", destroyMethod = "shutdown")
  40.     public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
  41.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroupName);
  42.         consumer.setNamesrvAddr(namesrvAddr);
  43.         consumer.setConsumeThreadMin(consumeThreadMin);
  44.         consumer.setConsumeThreadMax(consumeThreadMax);
  45.         consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
  46.         // 设置消息监听
  47.         consumer.registerMessageListener(orderlyListener);
  48.         // 设置消费起始位置位置
  49.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  50.         //设置消费模型，集群还是广播
  51.         consumer.setMessageModel(MessageModel.CLUSTERING);
  52.         consumer.subscribe(topicName, tag);
  53.         return consumer;
  54.     }
  55. }

四、生产及消费分区顺序消息

分区顺序消息监听器

  1. package com.zhangsan.boot.listener;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
  5. import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
  6. import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
  7. import org.apache.rocketmq.common.message.MessageExt;
  8. import org.springframework.stereotype.Component;
  9. import org.springframework.util.CollectionUtils;
  11. import java.util.List;
  13. /**
  14. * 分区顺序消费
  15. */
  16. @Component
  17. @Slf4j
  18. public class OrderlyListener implements MessageListenerOrderly {
  19.     @Override
  20.     public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
  21.         if (CollectionUtils.isEmpty(list)) {
  22.             log.info("MQ接收消息为空，直接返回成功");
  23.             return ConsumeOrderlyStatus.SUCCESS;
  24.         }
  25.         MessageExt messageExt = list.get(0);
  26.         try {
  27.             String topic = messageExt.getTopic();
  28.             String tags = messageExt.getTags();
  29.             int queueId = messageExt.getQueueId();
  30.             String body = new String(messageExt.getBody(), "utf-8");
  31.             log.info("MQ消息topic={},queueId={}, tags={}, 消息内容={}", topic,queueId,tags,body);
  32.         } catch (Exception e) {
  33.             log.error("获取MQ消息内容异常{}",e);
  34.             return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
  35.         }
  36.         // 处理业务逻辑
  37.         return ConsumeOrderlyStatus.SUCCESS;
  38.     }
  39. }

五、分区顺序消息生产

这里通过SelectMessageQueueByHash类实现orderId hash到同一queue中，可以看SelectMessageQueueByHash具体实现。

  1. package com.zhangsan.boot.impl;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  5. import org.apache.rocketmq.client.producer.SendResult;
  6. import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
  7. import org.apache.rocketmq.common.message.Message;
  8. import org.apache.rocketmq.remoting.common.RemotingHelper;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.beans.factory.annotation.Value;
  11. import org.springframework.stereotype.Service;
  13. import java.util.UUID;
  15. @Service("mQService")
  16. @Slf4j
  17. public class MQServiceImpl {
  19.     @Autowired
  20.     private DefaultMQProducer defaultMQProducer;
  22.     @Value("${spring.rocket.consumer.topicName}")
  23.     private String topic;
  25.     @Value("${spring.rocket.consumer.tag}")
  26.     private String tag;
  28.     public String sendMQMsg (){
  29.         for (int i = 0; i < 50; i++) {
  30.             //生成50个订单,这里订单号就用uuid来代替了
  31.             String orderId = UUID.randomUUID().toString();
  32.             // 每个订单有4个步骤,比如说 下单 支付 确认收货 ,评价 ,每个步骤都会发送一个消息过去,并且这个消息不允许顺序乱,也就是 不能 支付在下单之前过来
  33.             for (int j = 1; j <= 4; j++) {
  34.                 try {
  35.                     Message msg =new Message(topic, tag, "KEY" + orderId,
  36.                             ("订单Id:" + orderId + " 步骤:" + j).getBytes(RemotingHelper.DEFAULT_CHARSET));
  37.                     SendResult sendResult = defaultMQProducer.send(msg, new SelectMessageQueueByHash(), orderId);
  38.                 } catch (Exception e) {
  39.                     e.printStackTrace();
  40.                     log.info("消息发送失败,orderId:{}",orderId);
  41.                 }
  42.             }
  43.         }
  44.         return "success";
  45.     }
  47. }

六、编写测试方法及验证

  1. @Autowired
  2. private MQServiceImpl mqService;
  4. @GetMapping("/mqOrderLyTest")
  5. public String mqOrderLyTest(){
  6.     return mqService.sendMQMsg();
  7. }

1650703291(1).png
可以看到统一订单的不同步骤都在一个queue中，并且同一订单id消费都是按步骤有序消费的，这样就实现了分区顺序消息功能
1650703355(1).png
1650703578(1).png

七、生产及消费随机消息

这里通过不同的groupName，topic，tag来消费，所以需要改造原来文件

  1. server:
  2.   port: 8081
  3. spring:
  4.   datasource:
  5.     master:
  6.       url: jdbc:mysql://localhost:3306/pay?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&useLocalSessionState=true
  7.       username: root
  8.       password: 123456
  9.       driver-class-name: com.mysql.jdbc.Driver
  10.       type: com.alibaba.druid.pool.DruidDataSource
  11.     slave:
  12.       url: jdbc:mysql://localhost:3306/payslave?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true&useLocalSessionState=true
  13.       username: root
  14.       password: 123456
  15.       driver-class-name: com.mysql.jdbc.Driver
  16.       type: com.alibaba.druid.pool.DruidDataSource
  17.   redis:
  18.     host: localhost
  19.     port: 6379
  20.     database: 0
  21.     password: 123456
  22.     timeout: 3S
  23.   rocket:
  24.     #生产者
  25.     producer:
  26.       namesrvAddr: localhost:9876
  27.       producerGroupName: zhangsan_producer
  28.       # 消息最大长度 默认 1024 * 4 (4M)
  29.       maxMessageSize: 4096
  30.       # 发送消息超时时间，默认 3000
  31.       sendMsgTimeOut: 3000
  32.       retryTimesWhenSendFailed: 2
  33.     #消费者
  34.     consumer:
  35.       base:
  36.         namesrvAddr: localhost:9876
  37.         # 消费者线程数据量
  38.         consumeThreadMin: 5
  39.         consumeThreadMax: 32
  40.         # 设置一次消费消息的条数，默认1
  41.         consumeMessageBatchMaxSize: 1
  42.       orderly:
  43.         consumerGroupName: zhangsan_consumer
  44.         topicName: pay_success
  45.         tag: tag1
  46.       concurrently:
  47.         consumerGroupName: zhangsan_consumer1
  48.         topicName: pay_success1
  49.         tag: tag2
  52. logging:
  53.   level:
  54.     root: debug
  1. @Getter
  2. @Setter
  3. @Slf4j
  4. @Configuration
  5. @ConfigurationProperties(prefix = "spring.rocket.consumer.base")
  6. public class BaseConsumerConfig {
  7.     private String namesrvAddr;
  9.     // 消费者线程数据量
  10.     private Integer consumeThreadMin;
  11.     private Integer consumeThreadMax;
  12.     private Integer consumeMessageBatchMaxSize;
  13. }
  1. package com.zhangsan.boot.mq.rocketmq;
  3. import lombok.Getter;
  4. import lombok.Setter;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.springframework.boot.context.properties.ConfigurationProperties;
  7. import org.springframework.context.annotation.Configuration;
  9. @Getter
  10. @Setter
  11. @Slf4j
  12. @Configuration
  13. @ConfigurationProperties(prefix = "spring.rocket.consumer.orderly")
  14. public class OrderlyConsumerConfig {
  16.     private String consumerGroupName;
  18.     private String topicName;
  20.     private String tag;
  21. }
  1. package com.zhangsan.boot.mq.rocketmq;
  3. import lombok.Getter;
  4. import lombok.Setter;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.springframework.boot.context.properties.ConfigurationProperties;
  7. import org.springframework.context.annotation.Configuration;
  9. @Getter
  10. @Setter
  11. @Slf4j
  12. @Configuration
  13. @ConfigurationProperties(prefix = "spring.rocket.consumer.concurrently")
  14. public class ConcurrentlyConsumerConfig {
  16.     private String consumerGroupName;
  18.     private String topicName;
  20.     private String tag;
  22. }
  1. package com.zhangsan.boot.mq.rocketmq;
  3. import com.zhangsan.boot.listener.ConcurrentlyListener;
  4. import com.zhangsan.boot.listener.OrderlyListener;
  5. import lombok.Getter;
  6. import lombok.Setter;
  7. import lombok.extern.slf4j.Slf4j;
  8. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  9. import org.apache.rocketmq.client.exception.MQClientException;
  10. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  11. import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
  12. import org.springframework.beans.factory.annotation.Autowired;
  13. import org.springframework.context.annotation.Bean;
  14. import org.springframework.context.annotation.Configuration;
  16. @Getter
  17. @Setter
  18. @Slf4j
  19. @Configuration
  20. public class Consumer {
  22.     @Autowired
  23.     private BaseConsumerConfig baseConsumerConfig;
  25.     @Autowired
  26.     private OrderlyConsumerConfig orderlyConsumerConfig;
  28.     @Autowired
  29.     private ConcurrentlyConsumerConfig concurrentlyConsumerConfig;
  31.     @Autowired
  32.     private OrderlyListener orderlyListener;
  33.     @Autowired
  34.     private ConcurrentlyListener concurrentlyListener;
  36.     @Bean(initMethod = "start", destroyMethod = "shutdown")
  37.     public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
  38.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(orderlyConsumerConfig.getConsumerGroupName());
  39.         consumer.setNamesrvAddr(baseConsumerConfig.getNamesrvAddr());
  40.         consumer.setConsumeThreadMin(baseConsumerConfig.getConsumeThreadMin());
  41.         consumer.setConsumeThreadMax(baseConsumerConfig.getConsumeThreadMax());
  42.         consumer.setConsumeMessageBatchMaxSize(baseConsumerConfig.getConsumeMessageBatchMaxSize());
  43.         // 设置消息监听
  44.         consumer.registerMessageListener(orderlyListener);
  45.         // 设置消费起始位置位置
  46.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  48.         //设置消费模型，集群还是广播
  49.         consumer.setMessageModel(MessageModel.CLUSTERING);
  50.         consumer.subscribe(orderlyConsumerConfig.getTopicName(), orderlyConsumerConfig.getTag());
  51.         return consumer;
  52.     }
  54.     @Bean(initMethod = "start", destroyMethod = "shutdown")
  55.     public DefaultMQPushConsumer concurrentConsumer() throws MQClientException {
  56.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(concurrentlyConsumerConfig.getConsumerGroupName());
  57.         consumer.setNamesrvAddr(baseConsumerConfig.getNamesrvAddr());
  58.         consumer.setConsumeThreadMin(baseConsumerConfig.getConsumeThreadMin());
  59.         consumer.setConsumeThreadMax(baseConsumerConfig.getConsumeThreadMax());
  60.         consumer.setConsumeMessageBatchMaxSize(baseConsumerConfig.getConsumeMessageBatchMaxSize());
  61.         // 设置消息监听
  62.         consumer.registerMessageListener(concurrentlyListener);
  63.         // 设置消费起始位置位置
  64.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  65.         //设置消费模型，集群还是广播
  66.         consumer.setMessageModel(MessageModel.CLUSTERING);
  67.         consumer.subscribe(concurrentlyConsumerConfig.getTopicName(), concurrentlyConsumerConfig.getTag());
  68.         return consumer;
  69.     }
  70. }
  1. package com.zhangsan.boot.listener;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  5. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  6. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  7. import org.apache.rocketmq.common.message.MessageExt;
  8. import org.springframework.stereotype.Component;
  9. import org.springframework.util.CollectionUtils;
  11. import java.util.List;
  12. @Slf4j
  13. @Component
  14. public class ConcurrentlyListener implements MessageListenerConcurrently {
  16.     @Override
  17.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
  18.         if (CollectionUtils.isEmpty(list)) {
  19.             log.info("并发消费MQ接收消息为空，直接返回成功");
  20.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  21.         }
  22.         MessageExt messageExt = list.get(0);
  23.         try {
  24.             String topic = messageExt.getTopic();
  25.             String tags = messageExt.getTags();
  26.             int queueId = messageExt.getQueueId();
  27.             String body = new String(messageExt.getBody(), "utf-8");
  28.             log.info("并发消费MQ消息topic={},queueId={}, tags={}, 消息内容={}", topic,queueId,tags,body);
  29.         } catch (Exception e) {
  30.             log.error("并发消费获取MQ消息内容异常{}",e);
  31.         }
  32.         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  33.     }
  34. }
  1. package com.zhangsan.boot.impl;
  3. import com.zhangsan.boot.mq.rocketmq.ConcurrentlyConsumerConfig;
  4. import com.zhangsan.boot.mq.rocketmq.OrderlyConsumerConfig;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  7. import org.apache.rocketmq.client.producer.SendResult;
  8. import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
  9. import org.apache.rocketmq.common.message.Message;
  10. import org.apache.rocketmq.remoting.common.RemotingHelper;
  11. import org.springframework.beans.factory.annotation.Autowired;
  12. import org.springframework.stereotype.Service;
  14. import java.util.UUID;
  16. @Service("mQService")
  17. @Slf4j
  18. public class MQServiceImpl {
  20.     @Autowired
  21.     private DefaultMQProducer defaultMQProducer;
  23.     @Autowired
  24.     private OrderlyConsumerConfig orderlyConsumerConfig;
  26.     @Autowired
  27.     private ConcurrentlyConsumerConfig concurrentlyConsumerConfig;
  29.     public String sendMQMsg (){
  30.         for (int i = 0; i < 50; i++) {
  31.             //生成50个订单,这里订单号就用uuid来代替了
  32.             String orderId = UUID.randomUUID().toString();
  33.             // 每个订单有4个步骤,比如说 下单 支付 确认收货 ,评价 ,每个步骤都会发送一个消息过去,并且这个消息不允许顺序乱,也就是 不能 支付在下单之前过来
  34.             for (int j = 1; j <= 4; j++) {
  35.                 try {
  36.                     Message msg =new Message(orderlyConsumerConfig.getTopicName(), orderlyConsumerConfig.getTag(), "KEY" + orderId,
  37.                             ("订单Id:" + orderId + " 步骤:" + j).getBytes(RemotingHelper.DEFAULT_CHARSET));
  38.                     SendResult sendResult = defaultMQProducer.send(msg, new SelectMessageQueueByHash(), orderId);
  39.                 } catch (Exception e) {
  40.                     e.printStackTrace();
  41.                     log.info("消息发送失败,orderId:{}",orderId);
  42.                 }
  43.             }
  44.         }
  45.         return "success";
  46.     }
  48.     public String sendRandomMQMsg (){
  49.         for (int i = 0; i < 50; i++) {
  50.             //生成50个订单,这里订单号就用uuid来代替了
  51.             String orderId = UUID.randomUUID().toString();
  52.             // 每个订单有4个步骤,比如说 下单 支付 确认收货 ,评价 ,每个步骤都会发送一个消息过去,并且这个消息不允许顺序乱,也就是 不能 支付在下单之前过来
  53.             for (int j = 1; j <= 4; j++) {
  54.                 try {
  55.                     Message msg =new Message(concurrentlyConsumerConfig.getTopicName(), concurrentlyConsumerConfig.getTag(), "KEY" + orderId,
  56.                             ("订单Id:" + orderId + " 步骤:" + j).getBytes(RemotingHelper.DEFAULT_CHARSET));
  57.                     SendResult sendResult = defaultMQProducer.send(msg);
  58.                 } catch (Exception e) {
  59.                     e.printStackTrace();
  60.                     log.info("消息发送失败,orderId:{}",orderId);
  61.                 }
  62.             }
  63.         }
  64.         return "success";
  65.     }
  67. }
  1. @Autowired
  2.     private MQServiceImpl mqService;
  4.     @GetMapping("/mqOrderLyTest")
  5.     public String mqOrderLyTest(){
  6.         return mqService.sendMQMsg();
  7.     }
  9.     @GetMapping("/mqConcurrentTest")
  10.     public String mqConcurrentTest(){
  11.         return mqService.sendRandomMQMsg();
  12.     }

可以看到同一订单在不同队列中
1650706999(1).png