RocketMQ_高级特性之顺序消息_1-【官方】百战程序员_IT在线教育培训机构

顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。

下面用订单进行分区有序的示例。一个订单的顺序流程是：创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中，消费时，同一个OrderId获取到的肯定是同一个队列。


xxxxxxxxxx
# 创建主题，8写8读
[root@node1 ~]# mqadmin updateTopic -b localhost:10911 -n localhost:9876 -r 8 -t tp_demo_01 -w 8
# 删除主题的操作：
[root@node1 ~]# mqadmin deleteTopic -c DefaultCluster deleteTopic -nlocalhost:9876 -t tp_demo_07
# 主题描述
[root@node1 ~]# mqadmin topicStatus -n localhost:9876 -t tp_demo_07

顺序生产消息


xxxxxxxxxx
public class OrderProducer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("producer_grp_02");
        producer.setNamesrvAddr("192.168.139.128:9876");
        producer.start();
        // 获取指定主题的MQ列表
        final List<MessageQueue> messageQueues = producer.fetchPublishMessageQueues("tp_demo_11");

        Message message = null;
        MessageQueue messageQueue = null;
        for (int i = 0; i < 100; i++) {
            // 采用轮询的方式指定MQ，发送订单消息，保证同一个订单的消息按顺序
            // 发送到同一个MQ
            messageQueue = messageQueues.get(i % 8);

            message = new Message("tp_demo_02", ("hello rocketmq order create - " + i).getBytes());
            producer.send(message, messageQueue);
            message = new Message("tp_demo_02", ("hello rocketmq order pay - " + i).getBytes());
            producer.send(message, messageQueue);
            message = new Message("tp_demo_02", ("hello rocketmq order delivery - " + i).getBytes());
            producer.send(message, messageQueue);
        }

        producer.shutdown();
    }
}

顺序消费消息


xxxxxxxxxx
public class OrderConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_grp_02");
        consumer.setNamesrvAddr("192.168.139.128:9876");
        consumer.subscribe("tp_demo_02", "*");

        consumer.setConsumeThreadMin(1);
        consumer.setConsumeThreadMax(1);
        consumer.setPullBatchSize(1);
        consumer.setConsumeMessageBatchMaxSize(1);

        // 使用有序消息监听器
        consumer.setMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {

                for (MessageExt msg : msgs) {
                    System.out.println(
                            msg.getTopic() + "\t" +
                            msg.getQueueId() + "\t" +
                            new String(msg.getBody())
                    );
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
    }
}

全局有序


xxxxxxxxxx
# 创建主题，1写1读
[root@node1 ~]# mqadmin updateTopic -b node1:10911 -n localhost:9876 -r 1 -t tp_demo_02 -w 1
# 删除主题的操作：
[root@node1 ~]# mqadmin deleteTopic -c DefaultCluster deleteTopic -n localhost:9876 -t tp_demo_02
# 主题描述
[root@node1 ~]# mqadmin topicStatus -n localhost:9876 -t tp_demo_02

全局生产者


xxxxxxxxxx
public class GlobalOrderProducer {
  public static void main(String[] args) throws MQClientException,
RemotingException, InterruptedException, MQBrokerException {
    DefaultMQProducer producer = new
DefaultMQProducer("producer_grp_02");
    producer.setNamesrvAddr("192.168.139.128:9876");
    producer.start();
    Message message = null;
    for (int i = 0; i < 100; i++) {
      message = new Message("tp_demo_02", ("hello rocketmq" +
i).getBytes());
      producer.send(message);
   }
    producer.shutdown();
 }
}

全局消费者


xxxxxxxxxx
public class GlobalOrderConsumer {
  public static void main(String[] args) throws MQClientException {
    DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("consumer_grp_02");
    consumer.setNamesrvAddr("192.168.139.128:9876");
    consumer.subscribe("tp_demo_02", "*");
    consumer.setConsumeThreadMin(1);
    consumer.setConsumeThreadMax(1);
    consumer.setPullBatchSize(1);
    consumer.setConsumeMessageBatchMaxSize(1);
    
    consumer.setMessageListener(new MessageListenerOrderly() {
      @Override
      public ConsumeOrderlyStatus consumeMessage(List<MessageExt>
msgs, ConsumeOrderlyContext context) {
        for (MessageExt msg : msgs) {
          System.out.println(new String(msg.getBody()));
            }
        return ConsumeOrderlyStatus.SUCCESS;
     }
   });
    consumer.start();
 }
}

实时效果反馈

1.顺序消息支持哪种发送方式？

A push方式

B pull方式

C 同步方式

D 异步方式

答案

1=>C

RocketMQ_高级特性之事务消息_2 RocketMQ_高级特性之顺序消息_2

北京市昌平区回龙观镇南店村综合商业楼2楼226室