前两篇文章介绍了 RocketMQ 基本概念与搭建,现在以它与 SpringBoot 的结合来介绍其基本的用法。
<!-- RocketMQ -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.2</version>
</dependency>
注意:rocketmq-spring-boot-starter
要与RocketMQ
的版本一致。
application.yaml
文件配置如下:
server:
port: 9007
spring:
application:
name: rockmq-producer
rocketmq:
# NameServer地址
name-server: 192.168.0.17:9876
producer:
# 生产者组
group: producer-group
# 发送同步消息失败时,重试次数,默认是 2
retry-times-when-send-failed: 2
# 发送异步消息失败时,重试次数,默认是 2
retry-times-when-send-async-failed: 2
# 发送消息超时时间,默认是 3s
send-message-timeout: 3000
下面接口发送的为同步消息,即必须收到 RocketMQ 服务响应后才能进行下一步,否则一直阻塞。
@RequestMapping("/rocketmq")
@RestController
public class ProducerController {
@Autowired
private RocketMQTemplate rocketMQTemplate;
/**
* 发送同步消息
*/
@RequestMapping("/syncSend")
public void syncSend() {
// 第一个参数指定Topic与Tag,格式: `topicName:tags`
// 第二个参数,消息内容
SendResult sendResult = rocketMQTemplate.syncSend("topicClean:tagTest", "syncMessage");
System.out.println("发送同步消息结果:" + sendResult.toString());
}
}
消费者的依赖同上面的生产者一样,同样是写下 yaml 文件配置。
server:
port: 9008
spring:
application:
name: rockmq-consumer
rocketmq:
# NameServer地址
name-server: 192.168.0.17:9876
生产者发送消息到 broker
后,消费者通过监听的方式获取broker
发送过来的消息。实现监听需要实现 RocketMQListener
接口:
/**
* 消费者监听器
*/
@Component
@RocketMQMessageListener(
consumerGroup = "consumer-group", //消费者组
topic = "topicClean", //topic
selectorExpression = "tagTest || tagB" //tag,可以有多个
)
public class ConsumerListener implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
System.out.println("接收消息:" + message);
}
}
分析一下参数内容
topic 这个是必须指定的,否则没有消息来源。
consumerGroup 是消费者组,这个必须制定。一条消息只能被同一个消费者组里的一个消费者消费。
selectorExpression
是用于消息过滤的,我们在生产的时候定义了tag内容,消费者可以指定消费某些tag的消息,具体策略如下:
上面的@RocketMQMessageListener
注解的常用配置参数:
参数
类型
默认值
说明
consumerGroup
String
消费者组
topic
String
Topic
selectorType
SelectorType
SelectorType.TAG
使用TAG 或者SQL92选择消息,默认tag
selectorExpression
String
"*"
控制哪些消息可以选择
consumeMode
ConsumeMode
ConsumeMode.CONCURRENTLY
消费模式,并发接收还是顺序接收,默认并发模式
messageModel
MessageModel
MessageModel.CLUSTERING
消费模式,广播模式还是集群模式,默认集群模式
consumeThreadMax
int
64
最大消费线程数
consumeTimeout
long
15L
消费超时时间(一条消息可能会阻塞使用线程的最长时间(以分钟为单位))
nameServer
String
配置文件中读取:$
nameServer地址
accessKey
String
配置文件中读取:$
AK
secretKey
String
配置文件中读取:$
SK
accessChannel
String
$
maxReconsumeTimes
int
-1
最大消息重试次数
首先第一步启动刚刚编写好的生产者及消费者服务。
调用生产者发送消息的接口/rocketmq/syncSend
后,控制台返回结果 sendStatus=SEND_OK,表示消息成功发送到 broker
:
发送同步消息结果:SendResult [sendStatus=SEND_OK, msgId=7F000001178C18B4AAC288364E780000, offsetMsgId=7C471A0C00002A9F0000000000031086, messageQueue=MessageQueue [topic=topicClean, brokerName=broker-a, queueId=2], queueOffset=4]
查看 RocketMQ 控制台消息界面,也可以查询到刚刚发出来的消息:
那么消费者是否成功的消费到消息了呢?这个我们暂时不清楚。
查看消费者控制台,很完美,消费者接收到了生产者的消息:
接收消息:syncMessage
同样,也可以查看 RocketMQ
控制台消费者界面,上面我们确定的消费者组是consumer-group
,点击查看消费详情,是能够看到成功地消费到了消息:
生产者发送的消息成功被消费者消费,说明了基本的消息流程是没问题的。
上面我们发送的是同步消息,那这么说除了同步消息,还有其他哪几种消息阿?不了解,那我们就继续往下看。
上面发送的同步消息属于普通消息,普通消息就是 RocketMQ 中无特性的消息,包含了同步消息、异步消息、单步发送消息 3 种。
同步消息是指消息发送方发出一条消息后,会在收到服务端返回响应之后才发下一条消息的通讯方式。
流程如下:
应用场景:这种可靠性同步地发送方式应用场景非常广泛,例如重要通知邮件、报名短信通知、营销短信系统等。
示例代码:
@RequestMapping("/syncSend")
public void syncSend() {
// 第一个参数指定Topic与Tag,格式: `topicName:tags`
// 第二个参数,消息内容
SendResult sendResult = rocketMQTemplate.syncSend("topicClean:tagTest", "syncMessage");
System.out.println("发送同步消息结果:" + sendResult.toString());
}
异步消息是指发送方发出一条消息后,不等服务端返回响应,接着发送下一条消息的通讯方式。RocketMQ 异步发送,需要实现异步发送回调接口(SendCallback)。消息发送方在发送了一条消息后,不需要等待服务端响应即可发送第二条消息,发送方通过回调接口接收服务端响应,并处理响应结果。
流程如下:
应用场景:异步发送一般用于链路耗时较长,对响应时间较为敏感的业务场景,例如,视频上传后通知启动转码服务,转码完成后通知推送转码结果等。
示例代码:
@RequestMapping("/asyncSend")
public void asyncSend() {
rocketMQTemplate.asyncSend("topicClean:tagTest", "asyncMessage", new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("发送异步消息成功:" + sendResult.toString());
}
@Override
public void onException(Throwable throwable) {
System.out.println("发送异步消息失败:" + throwable.toString());
}
});
}
发送⽅只负责发送消息,不等待服务端返回响应且没有回调函数触发,即只发送请求不等待应答。
流程如下:
应用场景:需要极快的响应速度,但不能保证可靠性。
示例代码:
@RequestMapping("/oneWaySend")
public void oneWaySend() {
rocketMQTemplate.sendOneWay("topicClean:tagTest", "oneWayMessage");
}
顺序消息指的是,严格按照消息的发送顺序进行消费的消息(FIFO)。
默认情况下生产者会把消息以Round Robin轮询方式发送到不同的Queue分区队列;而消费消息时会从多个Queue上拉取消息,这种情况下的发送和消费是不能保证顺序的。
将消息仅发送到同一个Queue 中,消费时也只从这个 Queue 上拉取消息,就严格保证了消息的顺序性。
如何保证顺序:
顺序消息分为全局有序消息、分区有序消息两种。
当发送和消费参与的 Queue 只有一个时所保证的有序是整个 Topic 中消息的顺序, 称为全局顺序,因为一个 Topic 对应只有一个 Queue, 所以会严重影响性能。
流程如下:
在创建 Topic 时指定 Queue 的数量。有三种指定方式:
只要将 Queue 的数量设置为 1 便可实现消息的有序存储。
对于指定的一个 Topic,所有消息根据 hashKey 进行区块分区。 同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费。
在电商业务场景中,一个订单的流程是:创建、付款、推送、完成。在加入 RocketMQ 后,一个订单会分别产生对于这个订单的创建、付款、完成消息,如果我们把所有消息全部送入到 RocketMQ 中的一个主题中,这里该如何实现针对一个订单的消息顺序性呢!
流程如下:
要完成分区有序性,在生产者环节使用自定义的消息队列选择策略,确保订单号尾数相同的消息会被先后发送到同一个队列中(案例中主题有3个队列,生产环境中可设定成10个满足全部尾数的需求),然后再消费端开启负载均衡模式,最终确保一个消费者拿到的消息对于一个订单来说是有序的。
生产者示例代码如下:
首先创建 order
对象
public class Order {
private long orderId;
private String desc;
public long getOrderId() {
return orderId;
}
public Order setOrderId(long orderId) {
this.orderId = orderId;
return this;
}
public String getDesc() {
return desc;
}
public Order setDesc(String desc) {
this.desc = desc;
return this;
}
@Override
public String toString() {
return "Order{" +
"orderId=" + orderId +
", desc='" + desc + '\'' +
'}';
}
}
接着创建生产者分区顺序消息发送接口:
/**
* 发送分区顺序消息
*/
@RequestMapping("/syncSendOrderly")
public void syncSendOrderly() {
List<Order> orderList = new ArrayList<>();
Order order = new Order();
//订单1
order.setOrderId(001).setDesc("创建");
orderList.add(order);
order = new Order().setOrderId(001).setDesc("付款");
orderList.add(order);
order = new Order().setOrderId(001).setDesc("完成");
orderList.add(order);
//订单2
order = new Order().setOrderId(002).setDesc("创建");
orderList.add(order);
order = new Order().setOrderId(002).setDesc("付款");
orderList.add(order);
order = new Order().setOrderId(002).setDesc("完成");
orderList.add(order);
//订单3
order = new Order().setOrderId(003).setDesc("创建");
orderList.add(order);
order = new Order().setOrderId(003).setDesc("付款");
orderList.add(order);
order = new Order().setOrderId(003).setDesc("完成");
orderList.add(order);
for (int i = 0; i < orderList.size(); i++) {
//分区顺序消息
//以orderId作为hashKey,一个 orderId 只会发送到一个 queue
SendResult sendResult = rocketMQTemplate.syncSendOrderly(
"order-topic",
"orderId:" +orderList.get(i).getOrderId() + ",orderMessage" + i,
String.valueOf(orderList.get(i).getOrderId()));
System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
sendResult.getSendStatus(),
sendResult.getMessageQueue().getQueueId(),
orderList.get(i).toString()));
}
}
postman 调用接口测试,控制台输出结果如下:
结果显示相同的 orderId 被分配到同一个 queue,并且按照创建、付款、完成的步骤发送到了 broker,这里就实现了第一步:消息的有序存储。
顺序消费实际上有两个核心点,一个是生产者有序存储,另一个是消费者有序消费。
消费者示例代码如下:
/**
* 消费者顺序消费监听器
*/
@Component
@RocketMQMessageListener(
consumerGroup = "order--group", //消费者组
topic = "order-topic", //topic
consumeMode = ConsumeMode.ORDERLY //消费模式:顺序消费,默认为并发消费
)
public class OrderConsumerListener implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
System.out.println("receive order message:" + message);
}
}
注意:RocketMQMessageListener 的 consumeMode 属性默认为 ConsumeMode.CONCURRENTLY,实现顺序消息需要将类型需改为ConsumeMode.ORDERLY。
控制台显示消费结果如下:
可以看到相同 orderId 的消息对应内容也是有序的。
当消息写入到 broker 后,在指定的时长后才可被消费处理的消息,称为延时消息。
延时消息的延迟时长不支持随意时长的延迟,是通过特定的延迟等级来指定的。
messageDelayLevel = '1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h';
即,若指定的延时等级为 3,则表示延迟时长为 10s,即延迟等级是从 1 开始计数的。
当然,如果需要自定义的延时等级,可以通过在broker加载的 conf 配置中 新增如下配置(例如下面增加了 1 天这个等级 1d)。
messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d
延时消息实现原理如下:
主要包含以下6个步骤:
修改消息 Topic 名称和队列信息
RocketMQ Broker 端在存储生产者写入的消息时,首先都会将其写入到 CommitLog 中。之后根据消息中的 Topic 信息和队列信息,将其转发到目标 Topic 的指定队列(ConsumeQueue)中。
由于消息一旦存储到 ConsumeQueue 中,消费者就能消费到,而延迟消息不能被立即消费,所以这里将Topic的名称修改为 SCHEDULE_TOPIC_XXXX,并根据延迟级别确定要投递到哪个队列下。同时,还会将消息原来要发送到的目标 Topic 和队列信息存储到消息的属性中。
转发消息到延迟主题 SCHEDULE_TOPIC_XXXX 的 CosumeQueue 中
CommitLog 中的消息转发到 CosumeQueue中 是异步进行的。在转发过程中,会对延迟消息进行特殊处理,主要是计算这条延迟消息需要在什么时候进行投递。
投递时间 = 消息存储时间(storeTimestamp) + 延迟级别对应的时间
延迟服务消费 SCHEDULE_TOPIC_XXXX 消息
Broker 内部有一个 ScheduleMessageService 类,其充当延迟服务,主要是消费 SCHEDULE_TOPIC_XXXX 中的消息,并投递到目标 Topic 中。
ScheduleMessageService 在启动时,其会创建一个定时器 Timer,并根据延迟级别的个数,启动对应数量的 TimerTask,每个 TimerTask 负责一个延迟级别的消费与投递。
将信息重新存储到 CommitLog 中
在将消息到期后,需要投递到目标 Topic。由于在第一步已经记录了原来的 Topic 和队列信息,因此这里重新设置,再存储到 CommitLog 即可。
将消息投递到目标 Topic 中
消费者消费目标 Topic 中的数据。
应用场景: 在12306平台中,车票预订成功后就会发送一条延迟消息。这条消息将会在45分钟后投递给后台业务系统(Consumer),后台业务系统收到该消息后会判断对应的订单是否已经完成支付。如果未完成,则取消预订,将车票再次放回到票池;如果完成支付,则忽略。
示例代码如下:
/**
* 发送延时消息
*/
@RequestMapping("/delaySend")
public void delaySend() {
//发送超时=3s,延时等级=3,延迟10s消费
SendResult sendResult = rocketMQTemplate.syncSend("topicClean:tagTest",
MessageBuilder.withPayload("延迟10s消息").build(), 3000, 3);
System.out.println("发送延时消息:" + sendResult.toString());
}
RocketMQ 事务消息(Transactional Message)是指应用本地事务和发送消息操作可以被定义到全局事务中,要么同时成功,要么同时失败。RocketMQ 的事务消息提供类似 X/Open XA 的分布事务功能,通过事务消息能达到分布式事务的最终一致。
事务消息发送分为两个阶段。第一阶段会发送一个半事务消息,半事务消息是指暂不能投递的消息,生产者已经成功地将消息发送到了 Broker,但是 Broker 未收到生产者对该消息的二次确认,此时该消息被标记成“暂不能投递”状态,如果发送成功则执行本地事务,并根据本地事务执行成功与否,向 Broker 半事务消息状态(commit或者rollback),半事务消息只有 commit 状态才会真正向下游投递。如果由于网络闪断、生产者应用重启等原因,导致某条事务消息的二次确认丢失,Broker 端会通过扫描发现某条消息长期处于“半事务消息”时,需要主动向消息生产者询问该消息的最终状态(Commit或是Rollback)。这样最终保证了本地事务执行成功,下游就能收到消息,本地事务执行失败,下游就收不到消息。总而保证了上下游数据的一致性。
整个事务消息的详细交互流程如下图所示:
事务消息发送步骤如下:
RocketMQ Broker
。RocketMQ Broker
将消息持久化成功之后,向生产者返回 Ack 确认消息已经发送成功,此时消息暂不能投递,为半事务消息。应用场景:如用户发起转账后,交易状态短暂挂起,发送指令给银行,如果发起失败则不发送指令,发送成功后等待结果更新交易状态。
示例代码如下:
生产者:
private Logger logger = LoggerFactory.getLogger(getClass());
/**
* 检查本地事务的状态
*/
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
logger.info("start check Local rocketMQ transaction");
RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;
try {
String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
logger.info("check trans msg content:{}", jsonStr);
} catch (Exception e) {
//异常就回滚
resultState = RocketMQLocalTransactionState.ROLLBACK;
}
return resultState;
}
说明:发送事务消息采用的是 sendMessageInTransaction 方法,返回结果为 TransactionSendResult 对象,该对象中包含了事务发送的状态、本地事务执行的状态等。
生产者监听器
发送事务消息除了生产者和消费者以外,我们还需要创建生产者的消息监听器,来监听本地事务执行的状态和检查本地事务状态。
/**
* 事务消息监听器
*/
@RocketMQTransactionListener
public class TransactionMsgListener implements RocketMQLocalTransactionListener {
private Logger logger = LoggerFactory.getLogger(getClass());
/**
* 执行本地事务
*/
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
Object arg) {
logger.info("start invoke local rocketMQ transaction");
RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;
try {
//处理业务
String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
logger.info("invoke msg content:{}", jsonStr);
String UUID = (String) arg;
logger.info("UUID:" + UUID);
} catch (Exception e) {
logger.error("invoke local mq trans error", e);
resultState = RocketMQLocalTransactionState.UNKNOWN;
}
return resultState;
}
/**
* 检查本地事务的状态
*/
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
logger.info("start check Local rocketMQ transaction");
RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;
try {
String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
logger.info("check trans msg content:{}", jsonStr);
} catch (Exception e) {
//异常就回滚
resultState = RocketMQLocalTransactionState.ROLLBACK;
}
return resultState;
}
}
executeLocalTransaction
是半事务消息发送成功后,执行本地事务的方法,具体执行完本地事务后,可以在该方法中返回以下三种状态:
LocalTransactionState.COMMIT_MESSAGE
:提交事务,允许消费者消费该消息LocalTransactionState.ROLLBACK_MESSAGE
:回滚事务,消息将被丢弃不允许消费。LocalTransactionState.UNKNOW
:暂时无法判断状态,等待固定时间以后 Broker 端根据回查规则向生产者进行消息回查。checkLocalTransaction
是由于二次确认消息没有收到,Broker 端回查事务状态的方法。回查规则:本地事务执行完成后,若 Broker 端收到的本地事务返回状态为 LocalTransactionState.UNKNOW,或生产者应用退出导致本地事务未提交任何状态。则 Broker 端会向消息生产者发起事务回查,第一次回查后仍未获取到事务状态,则之后每隔一段时间会再次回查。
消费者
/**
* 消费者监听器
*/
@Component
@RocketMQMessageListener(
consumerGroup = "consumer-group", //消费者组
topic = "topicClean", //topic
selectorExpression = "tagTest || tagB" //tag
)
public class ConsumerListener implements RocketMQListener<String> {
@Override
public void onMessage(String message) {
System.out.println("接收消息:" + message);
}
}
说明:事务消息的消费者与普通的消费者没有区别。
测试
调用事务消息接口,控制台打印日志如下:
l.p.listen.TransactionMsgListener : start invoke local rocketMQ transaction
l.p.listen.TransactionMsgListener : invoke msg content:this is transactionMessage
l.p.listen.TransactionMsgListener : UUID:39030439-551f-407a-970b-a85f0671bfac
l.p.controller.ProducerController : sendStatus:SEND_OK,localTransactionState:COMMIT_MESSAGE
通过日志我们可以看出,执行的流程与上述的一致,执行成功后,消息执行成功返回的结果为 SEND_OK,本地事务执行的状态为 COMMIT_MESSAGE。
异常测试
这里将修改executeLocalTransaction
方法内容,当处理业务出现异常时,直接设置本地事务状态为ROLLBACK
。
/**
* 执行本地事务
*/
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
Object arg) {
logger.info("start invoke local rocketMQ transaction");
RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;
try {
//处理业务
String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
logger.info("invoke msg content:{}", jsonStr);
//抛出异常
int i = 1/0;
} catch (Exception e) {
logger.error("invoke local mq trans error", e);
//设置事务状态为回滚
resultState = RocketMQLocalTransactionState.ROLLBACK;
}
return resultState;
}
注意:
executeLocalTransaction
返回本地事务状态为UNKNOWN
,Broker 端会进行事务回查,而事务回查执行的就是checkLocalTransaction
方法。executeLocalTransaction
返回本地事务状态为ROLLBACK
,则直接丢弃准备发给消费者的消息,结束消息发送流程。查看控制台,日志打印结果如下:
l.p.listen.TransactionMsgListener : start invoke local rocketMQ transaction
l.p.listen.TransactionMsgListener : invoke msg content:this is transactionMessage
l.p.listen.TransactionMsgListener : invoke local mq trans error
l.p.controller.ProducerController : sendStatus:SEND_OK,localTransactionState:ROLLBACK_MESSAGE
通过日志可以看出消息执行成功返回的结果为 SEND_OK,本地事务执行的状态为 ROLLBACK_MESSAGE。
本文演示了 Springboot 项目下 RocketMQ 消息的发送及消费流程,由最基本的同步消息举例讲解延伸到顺序消息、延时消息及事务消息这几种不同的消息类型。
想了解有关 RocketMQ 的更多知识点,且听下回(肝有点疼)。
参考资料:
手机扫一扫
移动阅读更方便
你可能感兴趣的文章