Spring Boot集成消息队列

Apache RocketMQ是由开源的轻量级消息队列，于2017年正式成为Apache顶级项目。

在分布式消息队列中间件领域，最热门的项目是Kafka和RocketMQ：

  • Kafka是较早开源的”消息处理平台”，在写吞吐量上，有明显优势，更适合处理日志类消息。

  • RocketMQ借鉴了部分Kafka的设计思路，并对实时性、大分区数等方面进行了优化，较适合做为业务类的消息。

因此，本书选用RocketMQ做为业务类的消息队列。

安装并运行RocketMQ

RocketMQ的容器化比较落后，基本没有可用的镜像版本，我们采用手工单机部署的方式。

首先，下载最新版二进制文件，当前是4.9.1：

  1. wget https://dlcdn.apache.org/rocketmq/4.9.1/rocketmq-all-4.9.1-bin-release.zip

完成后，解压缩：

  1. unizp rocketmq-all-4.9.1-bin-release.zip

启动Name Server：

  1. nohup sh bin/mqnamesrv &
  2. tail -f ~/logs/rocketmqlogs/namesrv.log

最后启动Broker：

  1. nohup sh bin/mqbroker -n 127.0.0.1:9876 &
  2. tail -f ~/logs/rocketmqlogs/broker.log

如果启动成功，在上述两个日志中，会有如下的日志：

  1. 2021-10-12 4:30:02 INFO main - tls.client.keyPassword = null
  2. 2021-10-12 4:30:02 INFO main - tls.client.certPath = null
  3. 2021-10-12 4:30:02 INFO main - tls.client.authServer = false
  4. 2021-10-12 4:30:02 INFO main - tls.client.trustCertPath = null
  5. 2021-10-12 4:30:02 INFO main - Using JDK SSL provider
  6. 2021-10-12 4:30:03 INFO main - SSLContext created for server
  7. 2021-10-12 4:30:03 INFO main - Try to start service thread:FileWatchService started:false lastThread:null
  8. 2021-10-12 4:30:03 INFO NettyEventExecutor - NettyEventExecutor service started
  9. 2021-10-12 4:30:03 INFO FileWatchService - FileWatchService service started
  10. 2021-10-12 4:30:03 INFO main - The Name Server boot success. serializeType=JSON
  12. 2021-10-12 14:36:09 INFO brokerOutApi_thread_3 - register broker[0]to name server 127.0.0.1:9876 OK
  13. 2021-10-12 14:36:09 ERROR DiskCheckScheduledThread1 - Error when measuring disk space usage, file doesn't exist on this path: /Users/coder4/store/commitlog
  14. 2021-10-12 14:36:18 ERROR StoreScheduledThread1 - Error when measuring disk space usage, file doesn't exist on this path: /Users/coder4/store/commitlog
  15. 2021-10-12 14:36:19 ERROR DiskCheckScheduledThread1 - Error when measuring disk space usage, file doesn't exist on this path: /Users/coder4/store/commitlog

可以发现，NameServer是没有问题的，Broker报了一个”Error when measuring disk space usage”的错，这个是当前版本的Bug，不影响使用。

如果想退出服务，可以直接kill，或者执行：

  1. sh bin/mqshutdown broker
  3. sh bin/mqshutdown namesrv

RocketMQ架构简介

在集成RocketMQ之前，先介绍一下RocketMQ的基本架构：

  • NameServer：轻量级元信息服务，管理路由信息并提供对应的读写服务

  • Broker：支撑TOPIC和QUEUE的存储，支持Push和Pull两种协议，有容错、副本、故障恢复机制。

  • Producer：发布端服务，支持分布式部署，并向Broker集群发送

  • Consumer：消费端服务，同时支持Push和Pull协议。支持消费、广播、顺序消息等特性。

  • Topic：队列，用于区分不同消息。

  • Tag：同一个Topic下，可以设定不同Tag(例如前缀)，通过Tag来过滤消息，只保留自己感兴趣的。

在使用Producer和Consumer时，需要指定消费组(Consumer Group)，这是从Kafka中借鉴过来的机制。相同Consumer Group下的实例会共享同一个GroupId，会被认为是对等的、可负载均衡的。事件会随机分发给相同GroupId下的多个实例中。

在Spring Boot中集成RocketMQ

首先引入依赖：

  1. implementation 'org.apache.rocketmq:rocketmq-client:4.9.1'

接着，我们创建生产者的抽象基类：

  1. package com.coder4.homs.demo.server.mq;
  3. import com.fasterxml.jackson.databind.ObjectMapper;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  6. import org.apache.rocketmq.client.producer.SendCallback;
  7. import org.apache.rocketmq.client.producer.SendResult;
  8. import org.apache.rocketmq.common.message.Message;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11. import org.springframework.beans.factory.DisposableBean;
  13. import javax.annotation.PostConstruct;
  14. import java.nio.charset.StandardCharsets;
  16. /**
  17.  * @author coder4
  18.  */
  19. public abstract class BaseProducer<T> implements DisposableBean {
  21.     private final Logger LOG = LoggerFactory.getLogger(getClass());
  23.     abstract String getNamesrvAddr();
  25.     abstract String getProducerGroup();
  27.     abstract String getTopic();
  29.     abstract String getTag();
  31.     protected DefaultMQProducer producer;
  33.     private ObjectMapper objectMapper = new ObjectMapper();
  35.     public BaseProducer() {
  36.         producer = new
  37.                 DefaultMQProducer(getProducerGroup());
  38.     }
  40.     @PostConstruct
  41.     public void postConstruct() {
  42.         producer.setNamesrvAddr(getNamesrvAddr());
  43.         try {
  44.             producer.start();
  45.         } catch (MQClientException e) {
  46.             LOG.error("producer start exception", e);
  47.             throw new RuntimeException(e);
  48.         }
  49.     }
  51.     @Override
  52.     public void destroy() throws Exception {
  53.         producer.shutdown();
  54.     }
  56.     protected Message buildMessage(String payload) {
  57.         return new Message(getTopic(),
  58.                 getTag(),
  59.                 payload.getBytes(StandardCharsets.UTF_8)
  60.         );
  61.     }
  63.     public void publish(T payload) {
  64.         try {
  65.             String val = objectMapper.writeValueAsString(payload);
  66.             producer.send(buildMessage(val));
  67.             LOG.info("publish success, topic = {}, tag = {}, msg = {}", getTopic(), getTag(), val);
  68.         } catch (Exception e) {
  69.             LOG.error("publish exception", e);
  70.         }
  71.     }
  73.     public void publishAsync(T payload) {
  74.         try {
  75.             String val = objectMapper.writeValueAsString(payload);
  76.             producer.send(buildMessage(val), new SendCallback() {
  77.                 @Override
  78.                 public void onSuccess(SendResult sendResult) {
  79.                     LOG.info("publishAsync success, topic = {}, tag = {}, msg = {}", getTopic(), getTag(), val);
  80.                 }
  82.                 @Override
  83.                 public void onException(Throwable e) {
  84.                     LOG.error("publish async exception", e);
  85.                 }
  86.             });
  87.         } catch (Exception e) {
  88.             LOG.error("publishAsync exception", e);
  89.         }
  90.     }
  92. }

如上所示：

  • nameServr、topic、tag由子类组成

  • 我们在构造函数中，创建了Producer对象

  • postConstruct中：设定了NameServer地址，并启动producer

  • publish / publishAsync：发送消息，先根据topic和tag构造消息，然后调用同步 / 异步的接口发送。

  • destroy时，停止producer

接下来我们看下Consumer的基类：

  1. /**
  2.  * @(#)BaseConsumer.java, 10月 12, 2021.
  3.  * <p>
  4.  * Copyright 2021 coder4.com. All rights reserved.
  5.  * CODER4.COM PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
  6.  */
  7. package com.coder4.homs.demo.server.mq;
  9. import com.fasterxml.jackson.core.type.TypeReference;
  10. import com.fasterxml.jackson.databind.ObjectMapper;
  11. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  12. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  13. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  14. import org.apache.rocketmq.client.exception.MQClientException;
  15. import org.apache.rocketmq.common.message.MessageExt;
  16. import org.slf4j.Logger;
  17. import org.slf4j.LoggerFactory;
  18. import org.springframework.beans.factory.DisposableBean;
  19. import org.springframework.util.CollectionUtils;
  21. import javax.annotation.PostConstruct;
  23. /**
  24.  * @author coder4
  25.  */
  26. public abstract class BaseConsumer<T> implements DisposableBean {
  28.     protected final Logger LOG = LoggerFactory.getLogger(getClass());
  30.     private static final int DEFAULT_BATCH_SIZE = 1;
  32.     private static final int MAX_RETRY = 1024;
  34.     abstract String getNamesrvAddr();
  36.     abstract String getConsumerGroup();
  38.     abstract String getTopic();
  40.     abstract String getTag();
  42.     abstract Class<T> getClassT();
  44.     abstract boolean process(T msg);
  46.     private ObjectMapper objectMapper = new ObjectMapper();
  48.     protected DefaultMQPushConsumer consumer;
  50.     public BaseConsumer() {
  51.         consumer = new
  52.                 DefaultMQPushConsumer(getConsumerGroup());
  53.     }
  55.     @PostConstruct
  56.     public void postConstruct() {
  57.         consumer.setNamesrvAddr(getNamesrvAddr());
  58.         try {
  59.             consumer.subscribe(getTopic(), getTag());
  60.         } catch (MQClientException e) {
  61.             LOG.error("consumer subscribe exception", e);
  62.             throw new RuntimeException(e);
  63.         }
  64.         consumer.setConsumeMessageBatchMaxSize(DEFAULT_BATCH_SIZE);
  66.         consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
  67.             if (CollectionUtils.isEmpty(msgs)) {
  68.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  69.             }
  71.             if (msgs.size() != DEFAULT_BATCH_SIZE) {
  72.                 LOG.error("MessageListenerConcurrently callback msgs.size() != 1");
  73.             }
  75.             MessageExt msg = msgs.get(0);
  76.             if (msg.getReconsumeTimes() >= MAX_RETRY) {
  77.                 LOG.error("reconsume exceed max retry times");
  78.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  79.             }
  81.             try {
  82.                 if (process(objectMapper.readValue(new String(msg.getBody()), getClassT()))) {
  83.                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  84.                 } else {
  85.                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  86.                 }
  87.             } catch (Exception e) {
  88.                 LOG.error("process exception", e);
  89.                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  90.             }
  91.         });
  92.         try {
  93.             consumer.start();
  94.         } catch (MQClientException e) {
  95.             LOG.error("consumer start exception", e);
  96.             throw new RuntimeException(e);
  97.         }
  98.     }
  100.     @Override
  101.     public void destroy() throws Exception {
  102.         consumer.shutdown();
  103.     }
  104. }

与Producer类似，topic、tag、namesrv由子类指定。

  • postConstruct：订阅了对应topic和tag的消息，并设定回掉函数，这里设定每批次最多拉取1个消息，以最简化处理失败的情况，你可以根据实际情况做出调整。

  • 接受消息时，会调用子类的process进行处理，同时进行json的反序列化操作

接下来，我们来写一个Demo的生产者、消费者：

首先配置nameSrv：

  1. # rocketmq
  2. rocketmq.namesrv: 127.0.0.1:9876

接着，定义消息：

  1. @Data
  2. @Builder
  3. @NoArgsConstructor
  4. @AllArgsConstructor
  5. public class DemoMessage {
  7.     private String msg;
  9.     private long ts;
  10. }

然后是具体的Consumer和Producer：

  1. package com.coder4.homs.demo.server.mq;
  3. import org.springframework.beans.factory.annotation.Value;
  4. import org.springframework.stereotype.Service;
  6. /**
  7.  * @author coder4
  8.  */
  9. @Service
  10. public class DemoConsumer extends BaseConsumer<DemoMessage> {
  12.     @Value("${rocketmq.namesrv}")
  13.     private String namesrv;
  15.     @Override
  16.     String getNamesrvAddr() {
  17.         return namesrv;
  18.     }
  20.     @Override
  21.     String getConsumerGroup() {
  22.         return "demo-consumer";
  23.     }
  25.     @Override
  26.     String getTopic() {
  27.         return "demo";
  28.     }
  30.     @Override
  31.     String getTag() {
  32.         return "*";
  33.     }
  35.     @Override
  36.     Class<DemoMessage> getClassT() {
  37.         return DemoMessage.class;
  38.     }
  40.     @Override
  41.     boolean process(DemoMessage msg) {
  42.         LOG.info("process msg = {}", msg);
  43.         return true;
  44.     }
  45. }
  1. package com.coder4.homs.demo.server.mq;
  3. import org.springframework.beans.factory.annotation.Value;
  4. import org.springframework.stereotype.Service;
  6. /**
  7.  * @author coder4
  8.  */
  9. @Service
  10. public class DemoProducer extends BaseProducer<DemoMessage> {
  12.     @Value("${rocketmq.namesrv}")
  13.     private String namesrv;
  15.     @Override
  16.     String getNamesrvAddr() {
  17.         return namesrv;
  18.     }
  20.     @Override
  21.     String getProducerGroup() {
  22.         return "demo-producer";
  23.     }
  25.     @Override
  26.     String getTopic() {
  27.         return "demo";
  28.     }
  30.     @Override
  31.     String getTag() {
  32.         return "*";
  33.     }
  34. }

我们可以调用Producer发送一个消息，然后会收到如下的日志，说明消息已经被成功处理！

  1. 2021-10-12 8:01:37.340  INFO 6270 --- [MessageThread_1] c.c.homs.demo.server.mq.DemoConsumer     : process msg = DemoMessage(msg=123, ts=1634032897315)

由于篇幅所限，我们只实战了基础的消息收发，推荐你根据文档继续探索其他内容，包括：集群部署)、顺序消息)、广播消息)等内容。