RocketMQ（04）——发送顺序消息

Posted 2022-06-25 elim168

发送顺序消息

如果你的业务上对消息的发送和消费顺序有较高的需求，那么在发送消息的时候你需要把它们放到同一个消息队列中，因为只有同一个队列的消息才能确保消费的顺序性。下面代码我们在发送消息的时候，调用的是需要传递MessageQueueSelector的send()，该方法还可以传递一个额外的参数，其对应MessageQueueSelector的select()的最后一个参数。下面代码中我们一共发送了10条消息，从1开始算顺序为奇数的都放到第一个队列中，顺序为偶数的都放第二个队列中。所以最终第一个队列放了顺序号为1/3/5/7/9的消息，第二个队列中放了顺序号为2/4/6/8/10的消息。

@Test
public void testOrderSend() throws Exception 
  DefaultMQProducer producer = new DefaultMQProducer("group1");
  producer.setNamesrvAddr(this.nameServer);
  producer.start();
  for (int i=0; i<10; i++) 
    Message message = new Message("topic1", "tag3", (System.currentTimeMillis() + "---" + System.nanoTime() + "hello ordered message " + i).getBytes());
    SendResult sendResult = producer.send(message, new MessageQueueSelector() 
      @Override
      public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) 
        int index = (int) arg;
        //奇数放一个队列，偶数放一个队列
        return mqs.get(index % mqs.size() % 2);
      
    , i);
    Assert.assertTrue(sendResult.getSendStatus() == SendStatus.SEND_OK);
  
  producer.shutdown();

在消费的时候如果需要确保队列中的消息是按照顺序消费的，注册消息监听器时不能再选择并发消费的MessageListenerConcurrently，而需要选择按顺序消费的MessageListenerOrderly。按顺序消费时每个线程会锁定当前队列，只有一条消息消费完了才会释放锁，这样确保同一队列同时只能有一个线程消费一条消息。而并发消费时是不会一直锁队列的。有序消费时同一个队列里面的消息会按照顺序进行消费，但是它们可能被不同的线程消费。如消息的顺序是1/2/3/4/5/6，则按照顺序消费可以保证消息的消费顺序一定是1/2/3/4/5/6，但是消费它们的线程有可能是线程6/5/4/3/2/1。如果要保证有序的消费是在同一个线程完成的，则消费者线程只能有一个，可以通过setConsumeThreadMax()定义消费线程的最大数，可以通过setConsumeThreadMin设置消费者线程的最小数。下面的代码中就定义了按照顺序进行消息的消费。

@Test
public void testOrderConsume() throws Exception 
  DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group2");
  consumer.setNamesrvAddr(this.nameServer);
  consumer.subscribe("topic1", "tag3");
  consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);//默认值，订阅以前的消息将被忽略
  consumer.registerMessageListener(new MessageListenerOrderly() 
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) 
      System.out.println(Thread.currentThread().getName() + "消费消息：" + new String(msgs.get(0).getBody()));
      return ConsumeOrderlyStatus.SUCCESS;
    
  );
  consumer.start();
  TimeUnit.SECONDS.sleep(120);
  consumer.shutdown();

顺序消息消费时返回的ConsumeOrderlyStatus只能是SUCCESS和SUSPEND_CURRENT_QUEUE_A_MOMENT。SUCCESS表示消费成功，可以继续消费下一条，而SUSPEND_CURRENT_QUEUE_A_MOMENT表示消费失败，需要等待一下再继续消费。它不能像并发消费那样跳过消费失败的消息，因为那样就破坏了消息消费的顺序性。
（注：本文是基于RocketMQ4.5.0所写）