RocketMQ入坑指南（五）：SpringBoot集成RocketMQ和具体使用方式

前言

经过前面几部分的教程，大家应该已经对RocketMQ有了一个全面的认识，建议仔细阅读前几章的内容，可以更好的理解这次的内容，接下来，我们通过代码来演示一下SpringBoot如何集成并使用RocketMQ发送消息

一、SpringBoot集成RocketMQ

集成很简单，只需要导入RocketMQ的starter包即可

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.1</version>
</dependency>
// 工具类方便打印信息
<dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.8.12</version>
</dependency>

在配置文件中配置相关参数

生产者

# 应用服务 WEB 访问端口
server:port: 8088
rocketmq:
# 多个nameserver用分号隔开name-server: 192.168.0.8:9876#生产者producer:# 生产者组名group: producer_group# 异步发送消息的重试次数retry-times-when-send-async-failed: 5

消费者：

# 应用服务 WEB 访问端口
server:port: 8081
rocketmq:
# 多个nameserver用分号隔开name-server: 192.168.0.8:9876# 消费者consumer:# 组名group: consumer_group# 消息模式 集群模式message-model: CLUSTERING
spring:application:name: consumer

二、发送普通消息

1.同步发送消息

同步消息是指消息发送方发出一条消息后，会在收到服务端返回响应之后才发下一条消息的通讯方式。

2.异步发送消息

异步消息是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。RocketMQ 异步发送，需要实现异步发送回调接口（SendCallback）。消息发送方在发送了一条消息后，不需要等待服务端响应即可发送第二条消息，发送方通过回调接口接收服务端响应，并处理响应结果。

3.单步发送消息

发送⽅只负责发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不等待应答。

具体代码如下：

生产者：

import cn.hutool.core.date.DateUtil;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.Date;@RestController
@RequestMapping("/rocketmq")
public class ProducerController {@Resourceprivate RocketMQTemplate rocketMQTemplate;/*** 发送普通消息* @return*/@RequestMapping("/testNormalMessage")public String testNormalMessage() throws Exception{//发送10条消息for (int i = 1; i <= 10; i++) {String msg = "这是第"+i+"个订单";//普通同步消息rocketMQTemplate.syncSend("TopicOrder",MessageBuilder.withPayload("同步消息："+msg));//普通异步消息rocketMQTemplate.asyncSend("TopicOrder", MessageBuilder.withPayload("异步消息："+msg), new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("发送成功："+sendResult.toString());}@Overridepublic void onException(Throwable throwable) {System.out.println("发生异常："+throwable.toString());}});//单步发送rocketMQTemplate.sendOneWay("TopicOrder",MessageBuilder.withPayload("单步发送消息："+msg));Thread.sleep(1000);}return "";}}

消费者：

import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;@RocketMQMessageListener(topic = "TopicOrder",consumerGroup = "${rocketmq.consumer.group}",selectorExpression = "*")
@Component
public class ConsumerNormal implements RocketMQListener<MessageExt> {@Overridepublic void onMessage(MessageExt messageExt) {try {System.out.println("收到信息："+new String(messageExt.getBody()));} catch (Exception e) {throw new RuntimeException(e);}}
}

结果：

可以看到消费者收到了生产者发送的消息

三、发送顺序消息

生产者：

 /*** 发送顺序通知* @return*/@RequestMapping("/testSyncOrderMessage")public String testSyncOrderMessage() throws Exception{//发送10个订单，每个订单发送3条记录数据for (int i = 1; i <= 10; i++) {for (int j = 1; j <= 3; j++) {String msg = "这是第"+i+"个订单";switch (j){case 1:msg += ",用于生成订单";break;case 2:msg += ",用来生成支付记录";break;case 3:msg += ",用来生成物流记录";break;}//自定义队列选择器
//                rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
//                    @Override
//                    public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
//                        return list.get(o.hashCode() % 2);
//                    }
//                });SendResult topicOrderOrderly = rocketMQTemplate.syncSendOrderly("TopicOrder_Orderly", msg, i + "");System.out.println("发送信息："+topicOrderOrderly.toString());Thread.sleep(1000);}}return "";}

消费者：

@RocketMQMessageListener(topic = "TopicOrder_Orderly",consumerGroup = "orderly_group",consumeMode = ConsumeMode.ORDERLY)
@Component
public class ConsumerOrderly implements RocketMQListener<MessageExt> {@Overridepublic void onMessage(MessageExt messageExt) {try {System.out.println("收到信息:队列ID："+messageExt.getQueueId()+" 内容："+new String(messageExt.getBody()));} catch (Exception e) {throw new RuntimeException(e);}}
}

我们可以启动两个消费者，观察多个消费者消费的数据情况，是否相同队列的数据由同一个消费者进行处理

结果：

可以从结果看出，相同订单号的订单由一个消费者按顺序消费，默认4个队列，两个消费者各承担了两个队列的消息

四、发送延迟消息

延迟等级前面的教程中介绍过，下面贴出来帮大家回忆一下

代码如下：

生产者：

/*** 发送延迟消息* @return*/@RequestMapping("/testDelayMessage")public String testDelayMessage() throws Exception{//发送延迟消息SendResult sendResult = rocketMQTemplate.syncSend("TopicDelay", //TopicMessageBuilder.withPayload("这是一条延迟10S的消息").build(), //发送的消息主体3000, //timeout时间3   //延迟等级，对应16个默认延迟时间);System.out.println(sendResult.toString());System.out.println(DateUtil.formatTime(new Date()));return "";}

消费者：

@RocketMQMessageListener(topic = "TopicDelay",consumerGroup = "delay_group",selectorExpression = "*")
@Component
public class ConsumerDelay implements RocketMQListener<MessageExt> {@Overridepublic void onMessage(MessageExt messageExt) {try {System.out.println("收到信息"+messageExt.toString());System.out.println(DateUtil.formatTime(new Date()));} catch (Exception e) {throw new RuntimeException(e);}}
}

结果：

生产者：

消费者：

五、发送事务消息(重要)

事务消息的整体逻辑跟前面章节介绍的方式相同，现在发送3条模拟消息分别代表

1.模拟成功消息

2.模拟二次确认失败消息

3.模拟二次确认结果未知，主动查询消息之后返回成功

预想的结果应该是第一和第三种可以成功消费

下面是测试代码：

1.新增生产者监听类，用于二次确认事务处理，和提供主动查询逻辑

import cn.hutool.core.date.DateUtil;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import java.nio.charset.StandardCharsets;
import java.util.Date;@RocketMQTransactionListener
public class OrderTransMsgListener implements RocketMQLocalTransactionListener {// 事务消息共有三种状态，提交状态、回滚状态、中间状态：// RocketMQLocalTransactionState.COMMIT: 提交事务，它允许消费者消费此消息。// RocketMQLocalTransactionState.ROLLBACK: 回滚事务，它代表该消息将被删除，不允许被消费。// RocketMQLocalTransactionState.UNKNOWN: 中间状态，它代表需要检查消息队列来确定状态。@Overridepublic RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {if("commit".equals(o)){System.out.println("二次确认成功："+new String((byte[])message.getPayload(), StandardCharsets.UTF_8));return RocketMQLocalTransactionState.COMMIT;}if("rollback".equals(o)){System.out.println("二次确认失败："+new String((byte[])message.getPayload(), StandardCharsets.UTF_8));return RocketMQLocalTransactionState.ROLLBACK;}if("unknown".equals(o)){System.out.println("二次确认未知："+new String((byte[])message.getPayload(), StandardCharsets.UTF_8));return RocketMQLocalTransactionState.UNKNOWN;}return null;}/*** 检查本地事务的状态* 回查间隔时间：系统默认每隔60秒发起一次定时任务，对未提交的半事务消息进行回查，共持续12小时。* 第一次消息回查最快*/@Overridepublic RocketMQLocalTransactionState checkLocalTransaction(Message message) {System.out.println("收到事务查询回调的时间：" + DateUtil.formatDateTime(new Date()));System.out.println("信息内容："+new String((byte[])message.getPayload(), StandardCharsets.UTF_8));return RocketMQLocalTransactionState.COMMIT;}
}

2.生产者

/*** 发送事务通知* @return*/@RequestMapping("/testTransMessage")public String testTransMessage() throws Exception{//模拟正常提交消息SendResult sendResult1 = rocketMQTemplate.sendMessageInTransaction("TopicTrans",MessageBuilder.withPayload("这是正常的消息").build(),"commit");System.out.println(sendResult1.toString());//模拟需要回滚的消息SendResult sendResult2 = rocketMQTemplate.sendMessageInTransaction("TopicTrans",MessageBuilder.withPayload("这是错误的消息").build(),"rollback");System.out.println(sendResult2.toString());//模拟生产者发送成功，二次确认失败，broker主动查询成功的情况SendResult sendResult3 = rocketMQTemplate.sendMessageInTransaction("TopicTrans",MessageBuilder.withPayload("这是二次确认失败，主动查询成功的消息").build(),"unknown");System.out.println(sendResult3.toString());return "";}

消费者：

@RocketMQMessageListener(topic = "TopicTrans",consumerGroup = "trans_group")
@Component
public class ConsumerTrans implements RocketMQListener<MessageExt> {@Overridepublic void onMessage(MessageExt messageExt) {try {System.out.println("收到信息:"+new String(messageExt.getBody())+"时间："+ DateUtil.formatDateTime(new Date()));} catch (Exception e) {throw new RuntimeException(e);}}
}

然后我们执行发送信息，看一下结果：

生产者：