Spring Boot快速集成RocketMQ实战教程-技术圈

前言

RocketMQ是目前主流的消息中间件之一，并且自身就支持分布式功能。最初由阿里巴巴团队开发，并且经历过双十一等海量消息场景的考验，后捐赠给Apache开源基金会，这也是为什么我们经常听说RocketMQ是阿里巴巴的消息中间件，项目却在Apache的顶级项目中。

网络上通过SpringBoot集成RocketMQ的教程很多，但大多数都无法做到快速、通用的进行集成。本篇文章带大家快速完成基于Spring Boot的集成使用，同时针对一些集成过程中的概念和使用方法以实例进行讲解。

RocketMQ的部署

关于RocketMQ的部署，通常有单Master模式、多Master模式、多Master多Slave模式（异步复制或同步双写）等。

本文重点介绍RocketMQ的集成部分，就不再这里讲解如何部署Master的部署过程，读者学习时只需部署单机模式或基于Docker部署即可。

依赖集成

首先创建一个SpringBoot项目，为了方便通过浏览器访问测试，引入web对应的starter。

<parent>    <groupId>org.springframework.bootgroupId>    <artifactId>spring-boot-starter-parentartifactId>    <version>2.4.0version>    <relativePath/> parent><dependencies>    <dependency>        <groupId>org.springframework.bootgroupId>        <artifactId>spring-boot-starter-webartifactId>    dependency>dependencies>

上面的依赖以及可以完成一个基于SpringBoot的web项目了。下面需要集成RocketMQ的依赖。

在此步骤中有两个选择，一个就是直接引入RocketMQ的依赖，比如：

<dependency>    <groupId>org.apache.rocketmqgroupId>    <artifactId>rocketmq-clientartifactId>    <version>4.3.0version>dependency>

但此种方式需要进行大量的配置及实例化操作，并不能够达到快速集成、方便使用的目的。

这里我们采用RocketMQ官方提供的基于spring的集成。项目的源码及依赖使用位于GitHub上：https://github.com/apache/rocketmq-spring 。

在该项目的ReadMe中已经清晰的描述了如何引入依赖：

<dependency>    <groupId>org.apache.rocketmqgroupId>    <artifactId>rocketmq-spring-boot-starterartifactId>    <version>${RELEASE.VERSION}version>dependency>

我们只需按照说明，引入对应的依赖即可，这里采用2.1.1版本。因此，引入依赖文件如下：

<dependency>    <groupId>org.apache.rocketmqgroupId>    <artifactId>rocketmq-spring-boot-starterartifactId>    <version>2.1.1version>dependency>

引入依赖之后，剩下的就是配置文件的配置和使用了。

配置文件

我们知道，SpringBoot默认的starter内置了很多配置文件，直接通过yml文件进行配置即可使用。这里引入了rocketmq的starter，虽然并不是官方的，但使用方式基本一致。

在yml文件中配置如下参数：

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: test-group

name-server参数对应的就是部署的RocketMQ的Nameserver服务，如果有多个的话用英文分号（;）进行分割。

如果使用的是SpringBoot2.0+的框架或者是JDK10，可将name-server改成nameServer。否则，可能会出现一些奇怪的bug。

上面是简化了的最基础的配置，其他的配置均采用默认配置，如果需要定制化配置，可对具体参数按照统一形式进行配置。

生产者示例

当完成了上面的集成，生产者使用其实非常简单，只需要在使用的地方注入RocketMQTemplate对象，然后调用其对应的发送方法即可。简单示例如下：

@Componentpublic class TestSendService {
    @Resource    private RocketMQTemplate rocketMQTemplate;
    public void send() {        rocketMQTemplate.send("test-topic-1",                MessageBuilder.withPayload("Hello, World! I'm from spring message").build());    }}

但如果是在项目中，这样每次使用都注入一个RocketMQTemplate并不符合面向对象的思想，而且RocketMQTemplate还提供了多个常用的方法，比如同步、异步、直接发送等模式。

我们可以将其封装成为一个通用的Service，这样其他服务只需注入对应的Service，调用公共的方法即可，并且注明每个方法的使用场景。

抽象出来的Service接口如下：

/** * Rocket MQ 对应服务封装 * **/public interface RocketMqService {
    /**     * 同步发送消息
     *      * 当发送的消息很重要是，且对响应时间不敏感的时候采用sync方式;     *     * @param mqMsg 发送消息实体类     */    void send(MqMsg mqMsg);
    /**     * 异步发送消息，异步返回消息结果
     * 
     * 当发送的消息很重要，且对响应时间非常敏感的时候采用async方式；     *     * @param mqMsg 发送消息实体类     */    void asyncSend(MqMsg mqMsg);
    /**     * 直接发送发送消息，不关心返回结果，容易消息丢失，适合日志收集、不精确统计等消息发送;
     * 
     * 当发送的消息不重要时，采用one-way方式，以提高吞吐量；     *     * @param mqMsg 发送消息实体类     */    void syncSendOrderly(MqMsg mqMsg);
}

定义了不同类型消息发送的方法，同时在注释部分说明具体方法的使用场景。其中将发送的参数封装为MqMsg对象，MqMsg的结构如下：

public class MqMsg {
    /**     * 一级消息：消息topic     */    private String topic;
    /**     * 二级消息：消息topic对应的tags     */    private String tags;
    /**     * 消息内容     */    private String content;        // 省略getter/setter方法}

其中，topic为消息的主题，content为消息的内容，具体内容可根据生产者和消费者之间进行协定。

针对上述的接口，提供具体的方法实现：

@Service("rocketMqService")public class RocketMqServiceImpl implements RocketMqService {
    private static final Logger log = LoggerFactory.getLogger(RocketMqServiceImpl.class);
    @Resource    private RocketMQTemplate rocketMQTemplate;
    @Override    public void send(MqMsg mqMsg) {        log.info("send发送消息到mqMsg={}", mqMsg);        rocketMQTemplate.send(mqMsg.getTopic() + ":" + mqMsg.getTags(),                MessageBuilder.withPayload(mqMsg.getContent()).build());    }
    @Override    public void asyncSend(MqMsg mqMsg) {        log.info("asyncSend发送消息到mqMsg={}", mqMsg);        rocketMQTemplate.asyncSend(mqMsg.getTopic() + ":" + mqMsg.getTags(), mqMsg.getContent(),                new SendCallback() {                    @Override                    public void onSuccess(SendResult sendResult) {                        // 成功不做日志记录或处理                    }
                    @Override                    public void onException(Throwable throwable) {                        log.info("mqMsg={}消息发送失败", mqMsg);                    }                });    }
    @Override    public void syncSendOrderly(MqMsg mqMsg) {        log.info("syncSendOrderly发送消息到mqMsg={}", mqMsg);        rocketMQTemplate.sendOneWay(mqMsg.getTopic() + ":" + mqMsg.getTags(), mqMsg.getContent());    }}

其中异步发送方法asyncSend的异步返回结果中可以根据具体的业务场景进行针对性的处理。

上述方法中均默认使用了tag对topic进行分类。如果具体的业务中不需要tag，则可对上述方法中拼接的冒号+tag部分去除。本实例使用tag进行分类，方便用到时可以借鉴。

完成了上面的封装之后，在业务使用场景中只需注入对应的Service即可，后面测试时我们会进行演示。

消费者示例

关于消费者我们可以直接实现RocketMQListener接口，然后通过@RocketMQMessageListener注解来匹配目标消息。

这里为了演示统一topic下不同的tag的使用方法，分两个消费者来进行演示，直接看代码：

/** * 消息队列消费端使用示例 * **/@Service@RocketMQMessageListener(topic = MqTopicConstant.DEMO_TOPIC        , consumerGroup = MqTopicConstant.DEMO_CONSUMER_GROUP_REGISTERED        , selectorExpression = MqTopicConstant.DEMO_TAG_REGISTERED)public class MqRegisteredListenerDemo implements RocketMQListener<String> {
    private static final Logger log = LoggerFactory.getLogger(MqRegisteredListenerDemo.class);
    @Override    public void onMessage(String message) {        log.info("received registered message: {}", message);    }}

上面的消费者监听器是针对DEMO_TAG_REGISTERED（自定义变量）这个tag来进行处理的，如果不需要tag，可去掉selectorExpression元素配置即可。@RocketMQMessageListener注解里selectorExpression默认是*，也就是接收topic下全部消息。

上面用到了常量类MqTopicConstant，主要是统一定义对应的topic和tag，内容如下：

/** * 消息队列相关常亮配置，包括group、topic、tag * **/public class MqTopicConstant {
    /**     * 示例消息队列，topic1个     */    public static final String DEMO_TOPIC = "test-top-1";
    /**     * 注册tag     */    public static final String DEMO_TAG_REGISTERED = "registered";
    /**     * 修改tag     */    public static final String DEMO_TAG_MODIFY = "modify";
    /**     * consumer group     */    public static final String DEMO_CONSUMER_GROUP_REGISTERED = "consumer_test-top-1_registered";
    public static final String DEMO_CONSUMER_GROUP_MODIFY = "consumer_test-top-1_modify";
}

下面再来看看第二个监听“更新”功能的tag代码实现：

/** * 消息队列消费端使用示例 * **/@Service@RocketMQMessageListener(topic = MqTopicConstant.DEMO_TOPIC        , consumerGroup = MqTopicConstant.DEMO_CONSUMER_GROUP_MODIFY        , selectorExpression = MqTopicConstant.DEMO_TAG_MODIFY)public class MqModifyListenerDemo implements RocketMQListener<String> {
    private static final Logger log = LoggerFactory.getLogger(MqRegisteredListenerDemo.class);
    @Override    public void onMessage(String message) {        log.info("received modify message: {}", message);    }}

这个消费者与第一个不同的地方有两个，均在注解部分。其中topic两个是相同的，selectorExpression一个针对注册功能的tag进行过滤，一个针对修改信息的功能进行过滤。而消费组consumerGroup对应的值不能相同，否则启动时会抛出异常。

经过上述步骤，已经完成了生产者、消费者的配置。下面就同一个测试来验证服务的执行情况。

测试验证

定义一个Controller，用于外部请求，触发用户注册和修改操作的消息发送。关于测试也可以使用单元测试，但需要让线层阻塞，防止消费者还没接收到消息时，单元测试已经执行完成。

@RestController@RequestMapping("/demo")public class DemoController {
    @Resource    private RocketMqService rocketMqService;
    @GetMapping("/send")    public void send() {        MqMsg mqMsg = new MqMsg();        mqMsg.setTopic(MqTopicConstant.DEMO_TOPIC);        mqMsg.setTags(MqTopicConstant.DEMO_TAG_REGISTERED);
        // 此处可为其他VO对象，替换掉Map        Map<String, String> userInfo = new HashMap<>();        userInfo.put("username", "zhangsan");        userInfo.put("age", "12");        // 此处可封装为json等格式        mqMsg.setContent(userInfo.toString());        // 第一个发送注册消息        rocketMqService.asyncSend(mqMsg);
        mqMsg.setTags(MqTopicConstant.DEMO_TAG_MODIFY);        userInfo.put("age", "18");        // 此处可封装为json等格式        mqMsg.setContent(userInfo.toString());        // 发送修改消息        rocketMqService.asyncSend(mqMsg);    }}

在上述测试中第一部分发送了注册用户的消息，第二部分针对注册的消息进行了修改，又发送了一个消息。消息内容直接将Map转换为字符串了，在实战的过程中可根据双方协商，比如采用Json或其他序列化方法。

然后在浏览器访问：http://localhost:8080/demo/send ，触发消息的发送。

此时查看控制台，几乎在瞬间，就可以看到如下日志信息：

2020-11-24 19:31:17.900  INFO 92401 --- [nio-8080-exec-1] c.e.r.service.impl.RocketMqServiceImpl   : asyncSend发送消息到mqMsg=com.example.rocketmq.vo.MqMsg@7cda781f2020-11-24 19:31:17.906  INFO 92401 --- [nio-8080-exec-1] c.e.r.service.impl.RocketMqServiceImpl   : asyncSend发送消息到mqMsg=com.example.rocketmq.vo.MqMsg@7cda781f2020-11-24 19:31:17.942  INFO 92401 --- [MessageThread_1] c.e.r.listener.MqRegisteredListenerDemo  : received registered message: {age=12, username=zhangsan}2020-11-24 19:31:17.942  INFO 92401 --- [MessageThread_1] c.e.r.listener.MqRegisteredListenerDemo  : received modify message: {age=18, username=zhangsan}

很显然，消费者已经成功接收到消息，并且同一个topic，根据不同的tag分别进行处理。至此，一个完整是示例演示完毕。

小结

关于消息队列，还是其他很多方法都位于RocketMQTemplate当中，根据业务需要可以继续封装对应的Service。关于tag也还有不同的使用方式，大家可基于该文提供的基本框架和基本思路进一步完善。