RocketMQ（六）消息类型--顺序消息
站长
· 阅读数 17
rocketmq支持顺序消息，而在rocketmq-spring-boot-starter中，分别提供了顺序同步，顺序异步，顺序单向的方式。但是在阿里云平台提供的rocketmq中，文档中只支持同步顺序消息。
rocketmq-spring-boot-starter提供一下三种方法：
同步：syncSendOrderly
异步：asyncSendOrderly
单向：sendOneWayOrderly
##同步顺序消息实现 控制器：
/**
* 顺序消息同步发送
*/
@RequestMapping("/send/syncOrder")
public void syncOrder(){
rocketMqProducer.sendSyncOrder("顺序消息-顺序同步发送","test_syncOrder", UUID.randomUUID().toString());
}
生产者：
/**
* 发送顺序消息
*/
public void sendSyncOrder(String msgBody, String topic, String orderId) {
for (int i = 0; i < 10; i++) {
SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, MessageBuilder.withPayload(msgBody).build(), orderId);
if (ObjectUtils.isNotEmpty(sendResult)) {
//sendResult不空则表示消息发送成功
log.info("send success , send msg = {}, messageId = {}", msgBody, sendResult.getMsgId());
}
}
}
消费者：
/**
* RocketMqProducer
* @date: 2020/11/26
* @author weirx
* @version 3.0
*/
@Slf4j
@Component
@RocketMQMessageListener(
topic = "test_syncOrder",
selectorExpression = "*",
consumerGroup = "test_syncOrder",
consumeMode = ConsumeMode.ORDERLY)
public class SimpleSyncOrderMessageListener implements RocketMQListener<MessageExt> {
@Override
public void onMessage(MessageExt messageExt) {
byte[] body = messageExt.getBody();
String msg = new String(body);
log.info("receive sync message：{}", msg);
}
}
结果看到，消息发送且收到返回值的顺序与消费顺序是完全一致的：
2020-11-27 15:01:02.340 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E41E80000
2020-11-27 15:01:02.374 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E45040002
2020-11-27 15:01:02.403 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E45260004
2020-11-27 15:01:02.432 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E45450008
2020-11-27 15:01:02.451 INFO 16388 --- [MessageThread_1] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.451 INFO 16388 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E41E80000 cost: 0 ms
2020-11-27 15:01:02.453 INFO 16388 --- [MessageThread_1] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.453 INFO 16388 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E45040002 cost: 0 ms
2020-11-27 15:01:02.461 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E4561000B
2020-11-27 15:01:02.488 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E457D0013
2020-11-27 15:01:02.518 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E45990016
2020-11-27 15:01:02.539 INFO 16388 --- [MessageThread_2] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.539 INFO 16388 --- [MessageThread_2] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E45260004 cost: 0 ms
2020-11-27 15:01:02.555 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E45B60019
2020-11-27 15:01:02.567 INFO 16388 --- [MessageThread_3] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.567 INFO 16388 --- [MessageThread_3] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E45450008 cost: 0 ms
2020-11-27 15:01:02.584 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E45DC001F
2020-11-27 15:01:02.611 INFO 16388 --- [nio-8085-exec-1] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序同步发送, messageId = AC100208400418B4AAC2891E45F80025
2020-11-27 15:01:02.770 INFO 16388 --- [MessageThread_4] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.770 INFO 16388 --- [MessageThread_4] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E4561000B cost: 0 ms
2020-11-27 15:01:02.839 INFO 16388 --- [MessageThread_5] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.839 INFO 16388 --- [MessageThread_5] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E457D0013 cost: 0 ms
2020-11-27 15:01:02.839 INFO 16388 --- [MessageThread_5] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.839 INFO 16388 --- [MessageThread_5] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E45990016 cost: 0 ms
2020-11-27 15:01:02.871 INFO 16388 --- [MessageThread_6] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.871 INFO 16388 --- [MessageThread_6] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E45B60019 cost: 0 ms
2020-11-27 15:01:02.871 INFO 16388 --- [MessageThread_6] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.872 INFO 16388 --- [MessageThread_6] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E45DC001F cost: 1 ms
2020-11-27 15:01:02.872 INFO 16388 --- [MessageThread_6] c.b.m.r.c.SimpleSyncOrderMessageListener : receive sync message：顺序消息-顺序同步发送
2020-11-27 15:01:02.872 INFO 16388 --- [MessageThread_6] a.r.s.s.DefaultRocketMQListenerContainer : consume AC100208400418B4AAC2891E45F80025 cost: 0 ms
2020-11-27 15:01:16.690 INFO 16388 --- [d2-873c14ebc911] c.a.n.client.config.impl.ClientWorker : get changedGroupKeys:[]
##异步顺序消息实现 控制器：
/**
* 顺序消息异步发送
*/
@RequestMapping("/send/asyncOrder")
public void asyncOrder(){
rocketMqProducer.sendAsyncOrder("顺序消息-顺序异步发送","test_asyncOrder", UUID.randomUUID().toString());
}
生产者：
/**
* 发送异步顺序消息
*/
@SneakyThrows
public void sendAsyncOrder(String msgBody, String topic, String orderId) {
for (int i = 0; i < 10; i++) {
//这里为了清晰地展示每次发送消息的顺，添加一秒的间隔
Thread.sleep(1000);
rocketMQTemplate.asyncSendOrderly(topic, MessageBuilder.withPayload(msgBody).build(), orderId, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
log.info("send success , send msg = {}, messageId = {}", msgBody, sendResult.getMsgId());
}
@Override
public void onException(Throwable throwable) {
log.info("send failed , msg = {} ", throwable);
}
});
}
}
消费者：
/**
* RocketMqProducer
* @date: 2020/11/26
* @author weirx
* @version 3.0
*/
@Slf4j
@Component
@RocketMQMessageListener(
topic = "test_asyncOrder",
selectorExpression = "*",
consumerGroup = "test_asyncOrder",
consumeMode = ConsumeMode.ORDERLY)
public class SimpleAsyncOrderMessageListener implements RocketMQListener<MessageExt> {
@SneakyThrows
@Override
public void onMessage(MessageExt messageExt) {
byte[] body = messageExt.getBody();
String msg = new String(body);
log.info("receive sync message：{}", msg);
//为了明显的看见消费效果，时间有点长
Thread.sleep(15000);
}
}
结果看到，发送消息的id与消费顺序的id是一致的：
2020-11-27 14:40:08.810 INFO 34760 --- [ublicExecutor_5] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B22520002
2020-11-27 14:40:08.810 INFO 34760 --- [ublicExecutor_6] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B21550000
2020-11-27 14:40:08.818 INFO 34760 --- [MessageThread_1] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:40:09.301 INFO 34760 --- [ublicExecutor_8] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B26390007
2020-11-27 14:40:10.302 INFO 34760 --- [ublicExecutor_3] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B2A22000A
2020-11-27 14:40:11.305 INFO 34760 --- [ublicExecutor_4] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B2E0B000D
2020-11-27 14:40:12.302 INFO 34760 --- [ublicExecutor_6] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B31F30010
2020-11-27 14:40:13.305 INFO 34760 --- [ublicExecutor_8] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B35DC0013
2020-11-27 14:40:14.304 INFO 34760 --- [ublicExecutor_3] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B39C30016
2020-11-27 14:40:15.306 INFO 34760 --- [ublicExecutor_3] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B3DAC0019
2020-11-27 14:40:16.305 INFO 34760 --- [ublicExecutor_5] c.c.b.m.r.producer.RocketMqProducer : send success , send msg = 顺序消息-顺序异步发送, messageId = AC10020887C818B4AAC2890B4195001C
2020-11-27 14:40:23.820 INFO 34760 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B22520002 cost: 15002 ms
2020-11-27 14:40:23.821 INFO 34760 --- [MessageThread_1] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:40:38.821 INFO 34760 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B21550000 cost: 15000 ms
2020-11-27 14:40:38.821 INFO 34760 --- [MessageThread_1] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:40:53.821 INFO 34760 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B26390007 cost: 15000 ms
2020-11-27 14:40:53.821 INFO 34760 --- [MessageThread_1] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:41:08.821 INFO 34760 --- [MessageThread_1] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B2A22000A cost: 15000 ms
2020-11-27 14:41:08.835 INFO 34760 --- [MessageThread_2] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:41:23.836 INFO 34760 --- [MessageThread_2] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B2E0B000D cost: 15001 ms
2020-11-27 14:41:23.836 INFO 34760 --- [MessageThread_2] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:41:38.836 INFO 34760 --- [MessageThread_2] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B31F30010 cost: 15000 ms
2020-11-27 14:41:38.837 INFO 34760 --- [MessageThread_2] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:41:53.837 INFO 34760 --- [MessageThread_2] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B35DC0013 cost: 15000 ms
2020-11-27 14:41:53.837 INFO 34760 --- [MessageThread_2] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:42:08.838 INFO 34760 --- [MessageThread_2] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B39C30016 cost: 15001 ms
2020-11-27 14:42:08.850 INFO 34760 --- [MessageThread_3] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:42:23.851 INFO 34760 --- [MessageThread_3] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B3DAC0019 cost: 15001 ms
2020-11-27 14:42:23.851 INFO 34760 --- [MessageThread_3] .b.m.r.c.SimpleAsyncOrderMessageListener : receive sync message：顺序消息-顺序异步发送
2020-11-27 14:42:38.852 INFO 34760 --- [MessageThread_3] a.r.s.s.DefaultRocketMQListenerContainer : consume AC10020887C818B4AAC2890B4195001C cost: 15001 ms
注意： 消费者注意在注解增加consumeMode = ConsumeMode.ORDERLY属性表示是顺序消息：
##单向顺序消息实现 sendOneWayOrderly没有返回值，不做代码演示了。
##RocketMq如何实现顺序的？ 在rocketmq中是通过发消息时设置一个hashKey，通过对这个key做hash来将不同的消息划分到不同的队列当中，用于区分相同消息。下面通过源码看下具体实现： ####生产者发送消息源码，所有涉及流程从上到下排列：
/**
* @param hashKey 区分不同对列的key
**/
public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey) {
//继续后面的调用，增加了一个发送消息超时时间
return this.syncSendOrderly(destination, message, hashKey, (long)this.producer.getSendMsgTimeout());
}
public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey, long timeout) {
//消息不为空
if (!Objects.isNull(message) && !Objects.isNull(message.getPayload())) {
try {
long now = System.currentTimeMillis();
//封装成指定格式的消息
org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
//继续调用，多了一个消息队列选择器
//默认是hash的
//private MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash();
SendResult sendResult = this.producer.send(rocketMsg, this.messageQueueSelector, hashKey, timeout);
long costTime = System.currentTimeMillis() - now;
if (log.isDebugEnabled()) {
log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
}
return sendResult;
} catch (Exception var12) {
log.error("syncSendOrderly failed. destination:{}, message:{} ", destination, message);
throw new MessagingException(var12.getMessage(), var12);
}
} else {
log.error("syncSendOrderly failed. destination:{}, message is null ", destination);
throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
}
}
public SendResult send(Message msg, MessageQueueSelector selector, Object arg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
//去除topic空格
msg.setTopic(this.withNamespace(msg.getTopic()));
//调用默认生产者实现send方法
return this.defaultMQProducerImpl.send(msg, selector, arg, timeout);
}
public SendResult send(Message msg, MessageQueueSelector selector, Object arg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
//执行发送并选择队列的方法，同步，没有回调
return this.sendSelectImpl(msg, selector, arg, CommunicationMode.SYNC, (SendCallback)null, timeout);
}
private SendResult sendSelectImpl(Message msg, MessageQueueSelector selector, Object arg, CommunicationMode communicationMode, SendCallback sendCallback, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
long beginStartTime = System.currentTimeMillis();
//确定生产者服务状态
this.makeSureStateOK();
//校验消息内容
Validators.checkMessage(msg, this.defaultMQProducer);
//根据topic获取所有topic相关信息，这里的主角“队列”就被获取到了，默认一个topic是4个
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
MessageQueue mq = null;
try {
//获取到队列的信息，主要是用队列id
List<MessageQueue> messageQueueList = this.mQClientFactory.getMQAdminImpl().parsePublishMessageQueues(topicPublishInfo.getMessageQueueList());
Message userMessage = MessageAccessor.cloneMessage(msg);
String userTopic = NamespaceUtil.withoutNamespace(userMessage.getTopic(), this.mQClientFactory.getClientConfig().getNamespace());
userMessage.setTopic(userTopic);
//这里获取到具体要用哪一个队列，通过hash，然后与队列list的id对比
//主要是selector.select()方法，在messageQueueList.get(hashkey的hash值)
//public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
// int value = arg.hashCode();
// if (value < 0) {
// value = Math.abs(value);
// }
// value %= mqs.size();
// return (MessageQueue)mqs.get(value);
// }
mq = this.mQClientFactory.getClientConfig().queueWithNamespace(selector.select(messageQueueList, userMessage, arg));
} catch (Throwable var15) {
throw new MQClientException("select message queue throwed exception.", var15);
}
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeout < costTime) {
throw new RemotingTooMuchRequestException("sendSelectImpl call timeout");
} else if (mq != null) {
//拿到队列信息，继续发送步骤
return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, (TopicPublishInfo)null, timeout - costTime);
} else {
throw new MQClientException("select message queue return null.", (Throwable)null);
}
} else {
this.validateNameServerSetting();
throw new MQClientException("No route info for this topic, " + msg.getTopic(), (Throwable)null);
}
}
以上就是实现顺序消息发送的代码。
####消费者如何顺序消费？ 下面简单说说原理，源码较多，还没细看。
从消息发送时知道在一个topic中会有多个队列，那么消费时就存在以下两个问题： 1）如何保证一个队列只被一个消费者消费？ 2）如何保证一个消费者中只有一个线程能进行消费？
总结解决以上问题有两点： 1、创建消息拉取任务时，消息客户端向broker端申请锁定MessageQueue，使得一个MessageQueue同一个时刻只能被一个消费客户端消费 2、消息消费时，多线程针对同一个消息队列的消费先尝试使用synchronized申请独占锁，加锁成功才能进行消费，使得一个MessageQueue同一个时刻只能被一个消费客户端中一个线程消费