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.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.4.0</version>
    <relativePath/> 
</parent>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

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

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

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.3.0</version>
</dependency>

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

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

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

<!--add dependency in pom.xml-->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>${RELEASE.VERSION}</version>
</dependency>

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

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.1</version>
</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对象，然后调用其对应的发送方法即可。简单示例如下：

@Component
public 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 {

    /**
     * 同步发送消息<br/>
     * <p>
     * 当发送的消息很重要是，且对响应时间不敏感的时候采用sync方式;
     *
     * @param mqMsg 发送消息实体类
     */
    void send(MqMsg mqMsg);

    /**
     * 异步发送消息，异步返回消息结果<br/>
     * <p>
     * 当发送的消息很重要，且对响应时间非常敏感的时候采用async方式；
     *
     * @param mqMsg 发送消息实体类
     */
    void asyncSend(MqMsg mqMsg);

    /**
     * 直接发送发送消息，不关心返回结果，容易消息丢失，适合日志收集、不精确统计等消息发送;<br/>
     * <p>
     * 当发送的消息不重要时，采用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@7cda781f
2020-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@7cda781f
2020-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也还有不同的使用方式，大家可基于该文提供的基本框架和基本思路进一步完善。

---

关注公众号：程序新视界，一个让你软实力、硬技术同步提升的平台

除非注明，否则均为程序新视界原创文章，转载必须以链接形式标明本文链接

本文链接：http://www.choupangxia.com/2020/11/26/spring-boot-rocketmq/