RocketMQ 与 Spring Cloud 集成实战

一、环境准备

在开始集成 RocketMQ 与 Spring Cloud 之前，我们需要准备好相应的开发环境。

1. JDK 环境 确保你已经安装了 Java Development Kit（JDK），推荐使用 JDK 8 及以上版本。可以通过以下命令检查 JDK 是否安装成功：

java -version

如果显示 JDK 的版本信息，则说明安装成功。

2. Maven 环境 Maven 是 Java 项目的构建工具，用于管理项目的依赖和构建过程。同样通过命令检查是否安装：

mvn -version

若显示 Maven 的版本信息，则安装无误。如果未安装，可以从 Maven 官网下载并按照文档进行安装配置。

3. Spring Cloud 项目 创建一个 Spring Cloud 项目，可以使用 Spring Initializr（https://start.spring.io/）来快速生成项目骨架。在 Initializr 中，选择你需要的依赖，如 Spring Web 等基础依赖。下载生成的项目压缩包并解压到本地开发环境。

4. RocketMQ 环境 RocketMQ 有两种部署方式：单机模式和集群模式。对于开发测试，单机模式较为方便。你可以从 RocketMQ 官方 GitHub 仓库（https://github.com/apache/rocketmq）下载二进制包。解压后，按照官方文档进行单机模式的启动配置。启动 NameServer：

nohup sh bin/mqnamesrv &

启动 Broker：

nohup sh bin/mqbroker -n localhost:9876 &

通过以上步骤，基本的开发环境就准备完成了。

二、引入依赖

在 Spring Cloud 项目的 pom.xml 文件中引入 RocketMQ 相关依赖。

1. 引入 Spring Boot Starter for RocketMQ

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.0</version>
</dependency>

此依赖提供了 Spring Boot 与 RocketMQ 集成的基础支持，包括自动配置、消息发送和接收的相关功能。

2. 其他可能需要的依赖 根据项目具体需求，可能还需要引入一些其他依赖，例如：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

如果项目涉及 Web 相关功能，如通过 RESTful 接口触发消息发送等，就需要引入 spring-boot-starter-web 依赖。

三、配置 RocketMQ

在 Spring Cloud 项目的 application.properties 或 application.yml 文件中配置 RocketMQ 相关参数。

1. 使用 application.properties 配置

rocketmq.name-server=localhost:9876
rocketmq.producer.group=my-producer-group
rocketmq.consumer.group=my-consumer-group
rocketmq.consumer.topics=my-topic

• rocketmq.name-server：指定 NameServer 的地址，在单机模式下为 localhost:9876。
• rocketmq.producer.group：定义生产者组的名称，生产者组是一类生产者的集合，用于标识同一类生产者。
• rocketmq.consumer.group：定义消费者组的名称，消费者组是一类消费者的集合，这些消费者通常消费同一类消息且消费逻辑一致。
• rocketmq.consumer.topics：指定消费者订阅的主题名称。

2. 使用 application.yml 配置

rocketmq:
  name-server: localhost:9876
  producer:
    group: my-producer-group
  consumer:
    group: my-consumer-group
    topics: my-topic

两种配置方式本质上是一样的，只是格式不同，你可以根据个人习惯选择。

四、消息生产者

1. 创建生产者服务 在 Spring Cloud 项目中创建一个消息生产者服务类，例如 RocketMQProducerService。

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class RocketMQProducerService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String topic, Object message) {
        Message<?> msg = MessageBuilder.withPayload(message).build();
        rocketMQTemplate.send(topic, msg);
    }
}

在上述代码中：

• 首先通过 @Autowired 注入了 RocketMQTemplate，RocketMQTemplate 是 Spring Boot 集成 RocketMQ 后提供的用于发送消息的核心类。
• sendMessage 方法接收两个参数，一个是主题 topic，另一个是要发送的消息 message。通过 MessageBuilder 将消息构建为 Message 对象，然后使用 rocketMQTemplate.send 方法将消息发送到指定的主题。

2. 使用生产者服务发送消息 在控制器（Controller）或其他业务服务中调用 RocketMQProducerService 来发送消息。例如，创建一个 TestController：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class TestController {

    @Autowired
    private RocketMQProducerService rocketMQProducerService;

    @GetMapping("/sendMessage")
    public String sendMessage() {
        String topic = "my-topic";
        String message = "Hello, RocketMQ!";
        rocketMQProducerService.sendMessage(topic, message);
        return "Message sent successfully";
    }
}

在这个 TestController 中：

• 通过 @Autowired 注入了 RocketMQProducerService。
• sendMessage 方法定义了要发送的主题和消息内容，并调用 rocketMQProducerService.sendMessage 方法发送消息。最后返回一个表示消息发送成功的字符串。

五、消息消费者

1. 创建消费者服务 创建一个消息消费者服务类，例如 RocketMQConsumerService。

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

@Service
@RocketMQMessageListener(topic = "my-topic", consumerGroup = "my-consumer-group")
public class RocketMQConsumerService implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
        // 处理接收到的消息，这里简单打印，实际应用中进行具体业务逻辑处理
    }
}

在上述代码中：

• @Service 注解将该类标记为一个 Spring 服务组件。
• @RocketMQMessageListener 注解用于配置消费者监听的主题和消费者组。这里指定监听 my-topic 主题，并且属于 my-consumer-group 消费者组。
• 该类实现了 RocketMQListener 接口，并重写了 onMessage 方法，当消费者接收到消息时，会调用这个方法来处理消息。这里只是简单地将接收到的消息打印出来，在实际应用中，可以在这个方法中编写具体的业务逻辑，如更新数据库、调用其他服务等。

六、高级特性

1. 消息顺序性 RocketMQ 支持消息顺序性，在某些场景下，保证消息顺序非常重要，比如订单处理场景，订单创建、支付、发货等消息需要按照顺序处理。

• 生产者发送顺序消息 修改 RocketMQProducerService 类的 sendMessage 方法来发送顺序消息。

public void sendOrderlyMessage(String topic, Object message, String hashKey) {
    Message<?> msg = MessageBuilder.withPayload(message).build();
    rocketMQTemplate.sendOrderly(topic, msg, hashKey);
}

这里的 sendOrderly 方法接收三个参数，除了主题 topic 和消息 message 外，还需要一个 hashKey。hashKey 用于确定消息发送到哪个队列，相同 hashKey 的消息会被发送到同一个队列，从而保证顺序性。

• 消费者接收顺序消息 消费者端不需要额外配置，只要消费的是同一个队列的消息，就会按照顺序消费。但是在处理消息时要注意，因为是顺序消费，如果某个消息处理时间过长，会阻塞后续消息的消费。所以在 onMessage 方法中，要尽量保证消息处理的高效性。

2. 消息可靠性 RocketMQ 通过多种机制来保证消息的可靠性，如同步刷盘、异步刷盘、同步复制、异步复制等。

• 同步刷盘 在 Broker 的配置文件（broker.conf）中设置同步刷盘：

flushDiskType = SYNC_FLUSH

同步刷盘保证消息在写入内存后，立即刷写到磁盘，这样即使 Broker 宕机，也不会丢失未刷盘的消息。但是同步刷盘会降低消息写入的性能。

• 异步刷盘

flushDiskType = ASYNC_FLUSH

异步刷盘是将消息先写入内存，然后由后台线程异步刷盘到磁盘。这种方式性能较高，但在 Broker 宕机时，可能会丢失部分未刷盘的消息。

• 同步复制 在 Broker 配置文件中设置同步复制：

brokerRole = SYNC_MASTER

同步复制模式下，Master 节点会等待所有 Slave 节点都成功复制消息后，才返回成功给生产者，保证了消息的可靠性，但也会降低消息发送的性能。

• 异步复制

brokerRole = ASYNC_MASTER

异步复制模式下，Master 节点在将消息发送给 Slave 节点后，不需要等待 Slave 节点复制成功，就返回成功给生产者，性能较高，但在 Master 节点宕机时，可能会丢失部分未复制到 Slave 节点的消息。

3. 消息过滤 RocketMQ 支持消息过滤，消费者可以根据消息的属性或内容进行过滤，只接收符合条件的消息。

• 根据消息属性过滤 生产者在发送消息时设置消息属性：

public void sendMessageWithProperty(String topic, Object message) {
    Message<?> msg = MessageBuilder.withPayload(message)
          .setHeader("key1", "value1")
          .build();
    rocketMQTemplate.send(topic, msg);
}

消费者通过 @RocketMQMessageListener 注解的 selectorExpression 属性来设置过滤表达式：

@Service
@RocketMQMessageListener(topic = "my-topic", consumerGroup = "my-consumer-group", selectorExpression = "key1 = 'value1'")
public class RocketMQConsumerWithFilterService implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received filtered message: " + message);
    }
}

上述代码中，消费者只会接收属性 key1 的值为 value1 的消息。

• 根据消息内容过滤（SQL92 语法） 生产者发送消息同上述设置属性的方式。消费者设置过滤表达式时，可以使用更复杂的 SQL92 语法：

@Service
@RocketMQMessageListener(topic = "my-topic", consumerGroup = "my-consumer-group", selectorType = SelectorType.SQL92, selectorExpression = "age > 18 AND gender = 'male'")
public class RocketMQConsumerWithSQLFilterService implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received SQL filtered message: " + message);
    }
}

这里假设消息内容是包含 age 和 gender 字段的对象，消费者只会接收满足 age > 18 且 gender ='male' 条件的消息。需要注意的是，使用 SQL92 语法过滤消息时，Broker 需要开启相关配置，在 broker.conf 文件中添加：

enablePropertyFilter = true

七、集群部署

1. RocketMQ 集群部署 RocketMQ 集群通常由多个 NameServer 和多个 Broker 组成。

• NameServer 集群 NameServer 集群可以通过启动多个 NameServer 实例来实现。每个 NameServer 实例相互独立，不进行数据同步。在启动 NameServer 时，可以指定不同的端口：

nohup sh bin/mqnamesrv -p 9877 &
nohup sh bin/mqnamesrv -p 9878 &

在生产者和消费者配置中，需要指定多个 NameServer 地址，用分号分隔：

rocketmq.name-server=localhost:9876;localhost:9877;localhost:9878

• Broker 集群 Broker 集群分为单 Master 模式、多 Master 模式、多 Master 多 Slave 模式。
• 单 Master 模式 这种模式下只有一个 Master Broker，不建议在生产环境中使用，因为一旦 Master 宕机，整个集群不可用。
• 多 Master 模式 启动多个 Master Broker，每个 Master Broker 相互独立。在 broker.conf 文件中配置不同的 Broker 名称和端口等信息：

brokerName = broker-a
brokerId = 0
listenPort = 10911
namesrvAddr = localhost:9876;localhost:9877;localhost:9878

启动另一个 Master Broker：

brokerName = broker-b
brokerId = 0
listenPort = 10912
namesrvAddr = localhost:9876;localhost:9877;localhost:9878

生产者和消费者不需要特殊配置，RocketMQ 客户端会自动发现所有的 Master Broker。

• 多 Master 多 Slave 模式 这种模式下每个 Master Broker 都有对应的 Slave Broker。配置 Slave Broker 时，brokerId 设为非 0 值，例如：

brokerName = broker-a
brokerId = 1
listenPort = 11011
namesrvAddr = localhost:9876;localhost:9877;localhost:9878

在这种模式下，可以通过配置 brokerRole 来选择同步复制或异步复制。

2. Spring Cloud 项目在集群中的部署 Spring Cloud 项目在集群中部署时，每个实例都可以作为消息生产者和消费者。由于 RocketMQ 客户端会自动发现 NameServer 和 Broker 的地址，所以不需要对每个 Spring Cloud 实例进行额外的 RocketMQ 配置。但是要注意，在多个实例同时作为消费者时，可能会出现重复消费的问题。可以通过合理设置消费者组和消息幂等来解决这个问题。

• 消息幂等性 消息幂等性是指相同的消息多次消费产生的效果与一次消费的效果相同。在消费者端实现消息幂等性，可以通过数据库唯一约束、缓存等方式。例如，在处理消息前，先查询数据库或缓存，判断该消息是否已经处理过，如果已经处理过，则直接返回，不再重复处理。

八、常见问题及解决方法

1. 消息发送失败

• 原因分析
  • NameServer 地址配置错误，导致生产者无法连接到 NameServer。
  • 生产者组名称与 Broker 端配置不一致。
  • 网络问题，如防火墙阻挡了通信端口。
• 解决方法
  • 检查 application.properties 或 application.yml 中 rocketmq.name-server 的配置，确保地址正确。
  • 确认生产者组名称与 Broker 端配置一致，可以查看 Broker 的日志文件来确认。
  • 检查防火墙设置，确保 RocketMQ 通信端口（NameServer 默认 9876，Broker 默认 10911 等）未被阻挡。

2. 消息接收不到

• 原因分析
  • 消费者组名称与 Broker 端配置不一致。
  • 主题名称配置错误，消费者订阅的主题与生产者发送的主题不一致。
  • 消息过滤规则配置错误，导致符合条件的消息被过滤掉。
• 解决方法
  • 检查消费者组名称，确保与 Broker 端配置一致。
  • 仔细核对生产者发送的主题和消费者订阅的主题，保证两者一致。
  • 检查消息过滤规则，确保过滤规则符合实际需求。

3. 性能问题

• 原因分析
  • 同步刷盘和同步复制等配置会降低性能。
  • 消费者处理消息逻辑复杂，导致消费速度慢。
  • 网络带宽不足，影响消息的发送和接收速度。
• 解决方法
  • 根据业务需求调整刷盘和复制策略，如在对可靠性要求不是特别高的场景下，使用异步刷盘和异步复制。
  • 优化消费者的消息处理逻辑，提高处理效率，例如可以采用多线程处理等方式。
  • 检查网络带宽，确保网络环境满足业务需求，如有必要，升级网络带宽。

通过以上详细的步骤和内容，你可以成功地将 RocketMQ 与 Spring Cloud 进行集成，并根据业务需求进行各种配置和优化。在实际应用中，要根据具体的业务场景和性能要求，合理选择 RocketMQ 的特性和配置，以达到最佳的应用效果。