RocketMQ（六）消息类型--顺序消息

作者: 我犟不过你 | 来源:发表于2020-11-27 17:05 被阅读0次

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同一个时刻只能被一个消费客户端中一个线程消费

RocketMQ（六）消息类型--顺序消息

同步顺序消息实现

异步顺序消息实现

单向顺序消息实现

RocketMq如何实现顺序的？

生产者发送消息源码，所有涉及流程从上到下排列：

消费者如何顺序消费？

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java 核心技术

Java 进阶

JVM · Java虚拟机原理 · JVM上语言·框架· 生态系统

编程语言爱好者

java学习快到碗里来