【RocketMQ】SpringBoot整合RocketMQ

🎯 导读：本文档详细介绍了如何在Spring Boot应用中集成Apache RocketMQ，并实现消息生产和消费功能。首先通过创建消息生产者项目，配置POM文件引入RocketMQ依赖，实现同步消息发送，并展示了如何发送普通字符串消息、对象消息以及集合消息。接着，文档讲解了如何搭建消息消费者项目，包括配置RocketMQ监听器以消费不同类型的RocketMQ消息。此外，还探讨了RocketMQ的不同消息类型如延迟消息、顺序消息的发送方式，以及消息消费模式的选择与实现（负载均衡模式与广播模式）。

RocketMQ集成SpringBoot入门案例

搭建rocketmq-producer（消息生产者）

创建项目，完整的pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.3</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.test</groupId><artifactId>01-rocketmq-producer</artifactId><version>0.0.1-SNAPSHOT</version><name>rocketmq-producer</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- rocketmq的依赖 --><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.2</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin></plugins></build></project>

如果想用RocketMQ 5.x版本，可以用2.3.0

修改配置文件application.yml

注意rocketmq不是在spring层级下面的，很容易搞错，以为和SpringBoot整合就是在spring层级下面了

spring:application:name: rocketmq-producer
rocketmq:# rocketMq的nameServer地址name-server: 127.0.0.1:9876     producer:# 生产者组别group: test-group# 消息发送的超时时间send-message-timeout: 3000# 异步消息发送失败重试次数retry-times-when-send-async-failed: 2# 发送消息的最大大小，单位字节，这里等于4Mmax-message-size: 4194304      

在测试类里面测试发送消息

往test主题里面发送一个简单的字符串消息

/*** 注入rocketMQTemplate，我们使用它来操作mq*/
@Autowired
private RocketMQTemplate rocketMQTemplate;/*** 测试发送简单的消息** @throws Exception*/
@Test
public void testSimpleMsg() throws Exception {// 往test的主题里面发送一个简单的字符串消息// syncSend同步消息// asyncSend异步// 参数1:topic名 参数2：Object，数据SendResult sendResult = rocketMQTemplate.syncSend("test", "我是一个简单的消息");// 拿到消息的发送状态System.out.println(sendResult.getSendStatus());// 拿到消息的idSystem.out.println(sendResult.getMsgId());
}

运行后查看控制台

搭建rocketmq-consumer（消息消费者）

创建项目，完整的pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.3</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.test</groupId><artifactId>02-rocketmq-consumer</artifactId><version>0.0.1-SNAPSHOT</version><name>rocketmq-consumer</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- rocketmq的依赖 --><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.0.2</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin></plugins></build></project>

修改配置文件application.yml

spring:application:name: rocketmq-consumer
rocketmq:name-server: 127.0.0.1:9876
#    consumer：
#        group: aaa-group 不需要写，一个项目一般有很多消费者组

监听器SimpleMsgListener

消费者要消费消息，就添加一个监听器，SpringBoot一启动，监听器就开始持续工作
1、类上添加注解 @Component 和 @RocketMQMessageListener 。topic指定消费的主题，consumerGroup 指定消费组，一个主题可以有多个消费者组，一个消息可以被多个不同的组的消费者都消费
2、实现 RocketMQListener 接口，泛型可以为具体的数据类型，如果想拿到消息的其他参数（如消息头、消息体，例如key之类的），泛型用MessageExt

package com.test.listener;import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;@Component
@RocketMQMessageListener(topic = "test", consumerGroup = "test-group", messageModel = MessageModel.CLUSTERING)
public class SimpleMsgListener implements RocketMQListener<String> {/*** 消费消息的方法* * @param message 消息内容，类型和上面的泛型一致。如果泛型指定了固定的类型，消息体就是我们的参数*/@Overridepublic void onMessage(String message) {System.out.println(message);}
}

泛型使用MessageExt

@Component
@RocketMQMessageListener(topic = "test", consumerGroup = "test-group", messageModel = MessageModel.CLUSTERING)
public class SimpleMsgListener implements RocketMQListener<MessageExt> {/*** 消费消息的方法* * @param message 消息内容，类型和上面的泛型一致。如果泛型指定了固定的类型，消息体就是我们的参数*/@Overridepublic void onMessage(MessageExt message) {System.out.println(new String(message.getBody()));}
}

onMessage方法没有返回值，要表示消息签收？

• 方法报错就拒收
• 方法不报错就签收

启动rocketmq-consumer

查看控制台，发现我们已经监听到消息了

发送对象消息和集合消息

发送对象消息（消息内容为对象）

消费者监听的时候泛型中写对象的类型即可

生产者

rocketmq-producer 添加一个Order类

package com.test.domain;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;import java.util.Date;/*** 订单对象*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class Order {/*** 订单号*/private String orderId;/*** 订单名称*/private String orderName;/*** 订单价格*/private Double price;/*** 订单号创建时间*/private Date createTime;/*** 订单描述*/private String desc;}

rocketmq-producer 添加一个单元测试

/*** 测试发送对象消息** @throws Exception*/
@Test
public void testObjectMsg() throws Exception {Order order = new Order();order.setOrderId(UUID.randomUUID().toString());order.setOrderName("我的订单");order.setPrice(998D);order.setCreateTime(new Date());order.setDesc("加急配送");// 往test-obj主题发送一个订单对象rocketMQTemplate.syncSend("test-obj", order);
}

发送此消息

消费者

rocketmq-consumer 将Order类拷贝过来

rocketmq-consumer 添加一个监听器

package com.test.listener;import com.test.domain.Order;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;/*** 创建一个对象消息的监听* 1.类上添加注解@Component和@RocketMQMessageListener* 2.实现RocketMQListener接口，注意泛型的使用*/
@Component
@RocketMQMessageListener(topic = "test-obj", consumerGroup = "test-obj-group")
public class ObjMsgListener implements RocketMQListener<Order> {/*** 消费消息的方法** @param message*/@Overridepublic void onMessage(Order message) {System.out.println(message);}
}

重启rocketmq-consumer后查看控制台，监听的对象消息如下

发送集合消息（消息内容为集合）

• 生产者创建一个Order的集合，发送消息；
• 监听方修改泛型中的类型为 Object ，接收到消息之后，再做类型强转

发送不同模式的消息

发送同步消息

• 同步消息：消息由生产者发送到 broker 后，会得到一个确认，具有强可靠性。用于重要的消息通知、短信通知等

入门案例演示发送的就是同步消息。下面三种发送消息的方法，底层都是调用syncSend，发送的都是同步消息

• rocketMQTemplate.syncSend()
• rocketMQTemplate.send()
• rocketMQTemplate.convertAndSend()

发送异步消息

• rocketMQTemplate.asyncSend()

/*** 测试异步发送消息** @throws Exception*/
@Test
public void testAsyncSend() throws Exception {// 发送异步消息，发送完以后会有一个异步通知rocketMQTemplate.asyncSend("test", "发送一个异步消息", new SendCallback() {/*** 成功的回调* @param sendResult*/@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("发送成功");}/*** 失败的回调* @param throwable*/@Overridepublic void onException(Throwable throwable) {System.out.println("发送失败");}});// 测试一下异步的效果System.out.println("谁先执行");// 挂起jvm 不让方法结束System.in.read();
}

测试发现。“谁先执行”打印在前面，“发送成功” 打印在后面

发送单向消息

• 单向消息用在不关心发送结果的场景
• 吞吐量很大，存在消息丢失的风险，可用于日志信息的发送

/*** 测试单向消息** @throws Exception*/
@Test
public void testOnWay() throws Exception {// 发送单向消息，没有返回值和结果rocketMQTemplate.sendOneWay("test", "这是一个单向消息");
}

发送延迟消息

延迟等级，从1级开始分别对应：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

/*** 测试延迟消息** @throws Exception*/
@Test
public void testDelay() throws Exception {// 构建消息对象Message<String> message = MessageBuilder.withPayload("我是一个延迟消息").build();// 发送一个延时消息，延迟等级为4级，也就是30s后被监听消费// 参数3：连接MQ的超时时间 参数4：延迟等级SendResult sendResult = rocketMQTemplate.syncSend("test", message, 2000, 4);System.out.println(sendResult.getSendStatus());
}

运行后，查看消费者端，过了30s才被消费

发送顺序消息

上面消息的消费者都一样的实现方法。但顺序消息的消费不太一样，消费者需要单线程消费

生产者

/*** 测试顺序消费* mq会根据hash的值来存放到一个队列里面去** @throws Exception*/
@Test
public void testOrderly() throws Exception {List<Order> orders = Arrays.asList(new Order(UUID.randomUUID().toString().substring(0, 5), "张三的下订单", null, null, null, 1),new Order(UUID.randomUUID().toString().substring(0, 5), "张三的发短信", null, null, null, 1),new Order(UUID.randomUUID().toString().substring(0, 5), "张三的物流", null, null, null, 1),new Order(UUID.randomUUID().toString().substring(0, 5), "张三的签收", null, null, null, 1),new Order(UUID.randomUUID().toString().substring(0, 5), "李四的下订单", null, null, null, 2),new Order(UUID.randomUUID().toString().substring(0, 5), "李四的发短信", null, null, null, 2),new Order(UUID.randomUUID().toString().substring(0, 5), "李四的物流", null, null, null, 2),new Order(UUID.randomUUID().toString().substring(0, 5), "李四的签收", null, null, null, 2));// 我们控制流程为 下订单->发短信->物流->签收  hash的值为seq，也就是说 seq相同的会放在同一个队列里面，顺序消费orders.forEach(order -> {rocketMQTemplate.syncSendOrderly("test-obj", order, String.valueOf(order.getSeq()));});
}

运行发送消息

ObjMsgListener

consumeMode指定消费类型

• CONCURRENTLY 并发消费，不按照顺序
• ORDERLY 顺序消费（消息放到一个队列，用一个线程来消费）

package com.test.listener;import com.test.domain.Order;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;@Component
@RocketMQMessageListener(topic = "test-obj",consumerGroup = "test-obj-group",// 修改为顺序消费模式（单线程）consumeMode = ConsumeMode.ORDERLY
)
public class ObjMsgListener implements RocketMQListener<Order> {/*** 消费消息的方法* @param message*/@Overridepublic void onMessage(Order message) {System.out.println(message);}
}

重启rocketmq-consumer，查看控制台，消息按照我们的放入顺序进行消费了

发送事务消息

rocketmq-producer 添加一个单元测试

/*** 测试事务消息* 默认是sync（同步的）* 事务消息会有确认和回查机制* 事务消息都会走到同一个监听回调里面，所以我们需要使用tag或者key来区分过滤** @throws Exception*/
@Test
public void testTrans() throws Exception {// 构建消息体Message<String> message = MessageBuilder.withPayload("这是一个事务消息").build();// 发送事务消息（同步的） 最后一个参数才是消息主题TransactionSendResult transaction = rocketMQTemplate.sendMessageInTransaction("test", message, "消息的参数");// 拿到本地事务状态System.out.println(transaction.getLocalTransactionState());// 挂起jvm，因为事务的回查需要一些时间System.in.read();
}

rocketmq-producer 添加一个本地事务消息的监听（半消息）

/*** 事务消息的监听与回查* 类上添加注解@RocketMQTransactionListener 表示这个类是本地事务消息的监听类* 实现RocketMQLocalTransactionListener接口* 两个方法为执行本地事务，与回查本地事务*/
@Component
@RocketMQTransactionListener(corePoolSize = 4,maximumPoolSize = 8)
public class TmMsgListener implements RocketMQLocalTransactionListener {/*** 执行本地事务，这里可以执行想做的业务，比如操作数据库*     操作成功就 return RocketMQLocalTransactionState.COMMIT;*     操作失败就 return RocketMQLocalTransactionState.UNKNOWN;* 可以使用try catch来控制成功或者失败* @param msg* @param arg* @return*/@Overridepublic RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {// 拿到消息参数System.out.println(arg);// 拿到消息头System.out.println(msg.getHeaders());// 执行业务// 返回状态 COMMIT 或者 UNKNOWNif (成功){return RocketMQLocalTransactionState.COMMIT;}else{return RocketMQLocalTransactionState.UNKNOWN;}        }/*** 此方法为回查方法，执行需要等待一会 * 回查本地事务，只有上面的执行方法返回UNKNOWN时，才执行下面的方法 默认是1min回查* 这里可以执行一些检查的方法* 如果返回COMMIT，那么本地事务就算是提交成功了，消息就会被消费者看到** @param msg* @return*/@Overridepublic RocketMQLocalTransactionState checkLocalTransaction(Message msg) {System.out.println(msg);return RocketMQLocalTransactionState.COMMIT;}
}

事务消息的处理逻辑

1. 消息会先到事务监听类的执行方法
2. 如果返回状态为COMMIT，则消费者可以直接监听到
3. 如果返回状态为ROLLBACK，则消息发送失败，直接回滚
4. 如果返回状态为UNKNOW，则过一段时间走回查方法
5. 如果回查方法返回状态为UNKNOW或者ROLLBACK，则消息发送失败，直接回滚
6. 如果回查方法返回状态为COMMIT，则消费者可以直接监听到

消息过滤

tag过滤（常在消费者端过滤）

• 从源码注释得知，tag带在主题后面用:来携带tag

• 在org.apache.rocketmq.spring.support.RocketMQUtil 的getAndWrapMessage方法里面看到了具体细节，keys要在消息头里面携带

生产者

/*** 发送一个带tag的消息** @throws Exception*/
@Test
public void testTagMsg() throws Exception {// 发送一个tag为java的数据rocketMQTemplate.syncSend("test-tag:java", "我是一个带tag的消息");
}

TagMsgListener

1、类上添加注解@Component和@RocketMQMessageListener。selectorType = SelectorType.TAG, 指定使用tag过滤。 (也可以使用sql92 需要在配置文件broker.conf中开启enbalePropertyFilter=true)

2、selectorExpression = “java” 表达式，默认是*，支持"tag1 || tag2 || tag3" 监听多个标签 3、实现RocketMQListener接口，注意泛型的使用

注意：一个标签，用一个消费者组！！！
package com.test.listener;import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;@Component
@RocketMQMessageListener(topic = "test-tag",consumerGroup = "test-tag-group",// 注明使用tag模式selectorType = SelectorType.TAG,selectorExpression = "java||C"
)
public class TagMsgListener implements RocketMQListener<String> {/*** 消费消息的方法** @param message*/@Overridepublic void onMessage(String message) {System.out.println(message);}
}

sql92表达式

一般不用这种模式

发送消息的时候携带数字标签

监听的消息的时候，通过数字来过滤消息

表达式可用语法：

• AND, OR
• >, >=, <, <=, =
• BETWEEN A AND B, equals to >=A AND <=B
• NOT BETWEEN A AND B, equals to >B OR <A
• IN (‘a’, ‘b’), equals to =‘a’ OR =‘b’, this operation only support String type.
• IS NULL, IS NOT NULL, check parameter whether is null, or not.
• =TRUE, =FALSE, check parameter whether is true, or false.

Example: (a > 10 AND a < 100) OR (b IS NOT NULL AND b=TRUE)

Key过滤（可以在事务监听的类里面区分）

生产者

key需要放在消息头上面

/*** 发送一个带 key 的消息，我们使用事务消息，打断点查看消息头** @throws Exception*/
@Test
public void testKeyMsg() throws Exception {// 发送一个key为spring的事务消息Message<String> message = MessageBuilder.withPayload("我是一个带key的消息").setHeader(RocketMQHeaders.KEYS, "spring").build();rocketMQTemplate.sendMessageInTransaction("test", message, "我是一个带key的消息");
}

断点发送这个消息，查看事务里面消息头

我们在mq的控制台也可以看到

两种消费模式

RocketMQ消息消费的模式分为两种：负载均衡模式和广播模式

• CLUSTERING：负载均衡模式，多个消费者交替消费同一个主题里面的消息
• BROADCASTING：广播模式，每个消费者都消费一遍订阅的主题的消息

消费者1

再搭建一个消费者rocketmq-consumer-b，依赖和配置文件和rocketmq-consumer一致，记住端口修改一下，避免占用。rocketmq-consumer-b添加一个监听

package com.test.listener;import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;/*** messageModel  指定消息消费的模式*      CLUSTERING 为负载均衡模式*      BROADCASTING 为广播模式*/
@Component
@RocketMQMessageListener(topic = "test",consumerGroup = "test-group",// 集群模式messageModel = MessageModel.CLUSTERING
)
public class ConsumerBListener implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println(message);}
}

消费者2

修改rocketmq-consumer的SimpleMsgListener

/*** 创建一个简单消息的监听* 1.类上添加注解@Component和@RocketMQMessageListener** @RocketMQMessageListener(topic = "test", consumerGroup = "test-group")* topic指定消费的主题，consumerGroup指定消费组,一个主题可以有多个消费者组,一个消息可以被多个不同的组的消费者都消费* 2.实现RocketMQListener接口，注意泛型的使用*/
@Component
@RocketMQMessageListener(topic = "test", consumerGroup = "test-group",messageModel = MessageModel.CLUSTERING)
public class SimpleMsgListener implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {System.out.println(new Date());System.out.println(message);}
}

启动两个消费者

两个消费者是一个集群，且是同一个消费者组。当然，两个消费者可以写在一个项目中，写两个监听器就行

在生产者里面添加一个单元测试并且运行

/*** 测试消息消费的模式** @throws Exception*/
@Test
public void testMsgModel() throws Exception {for (int i = 0; i < 10; i++) {rocketMQTemplate.syncSend("test", "我是消息" + i);}
}

查看两个消费者的控制台

发现是负载均衡的模式，消息被负载均衡到两个消费者上，负载均衡不是说数量一定一样

BROADCASTING 广播模式

@RocketMQMessageListener(topic = "test",consumerGroup = "test-group",// 集群模式messageModel = MessageModel.BROADCASTING
)

重启测试，广播模式下，每个消费者都消费了这些消息

项目中一般部署多台机器，消费者部署 2 到 3 个，根据业务可以选择具体的模式来配置。广播模式是不会更新消费者位点的，它在乎消费失败，也不会重试，就广播一次