RocketMQ（06）——消息的批量发送和消费

Posted 2022-06-25 elim168

消息的批量发送和消费

发送

生产者进行消息发送时可以一次发送多条消息，这对于体积比较小的消息而言会大大改善性能，可以想象原本发送10条消息需要传递10次，现在只需要传递一次。这可以通过调用send()时传递一个消息集合。比如下面这样。批量发送的消息必须拥有相同的Topic，且一次发送的消息的body的总体积不能超过4MB，否则会得到异常——org.apache.rocketmq.client.exception.MQClientException: CODE: 13 DESC: the message body size over max value, MAX: 4194304，这是Broker端对消息大小的限制。可以通过setMaxMessageSize()指定生产者发送消息的大小限制，单位是字节，默认是4MB。如果需要发送超过4MB大小的消息，除了通过生产者的setMaxMessageSize()调整生产者这端的消息大小，还需要在Broker端通过maxMessageSize指定Broker允许的最大消息大小。此外，如果在批量发送消息的时候集合中包含的消息体的总体积超过4MB的，还可以对集合进行拆分，分为多次发送。

@Test
public void testSendBatch() throws Exception 
  DefaultMQProducer producer = new DefaultMQProducer("group1");
  producer.setNamesrvAddr(this.nameServer);
  producer.setMaxMessageSize(10 * 1024 * 1024);
  producer.start();
  String topic = "topic1";
  String tag = "tag6";
  List<Message> messages = new ArrayList<>();
  for (int i=0; i<3; i++) 
    messages.add(new Message(topic, tag, new byte[1024 * 1024]));
  
  //批量发送消息，一次发送的消息总量不能超过4MB。
  producer.send(messages);
  producer.shutdown();

消费

我们前面介绍的消息监听器的回调方法consumeMessage()的消息参数msgs是一个集合，但是默认情况下它每次都只包含一条消息。如果需要回调时一次传递多条消息，可以通过消费者的setConsumeMessageBatchMaxSize()指定一次最多消费的消息数量，默认是1。可能你想着我一次消费的数量越多越好，那我就定义一次消费50条。当定义了消息的最大消费数量是50时，实际上一次可消费的消息数量最多也就32条。因为Consumer默认一次最多只能拉取32条消息，可以通过setPullBatchSize()指定一次可以拉取消息的数量。根据笔者的测试，pullBatchSize小于32时，实际拉取的消息数量就是指定的拉取数量，但是当pullBatchSize大于32时，实际拉取回来的消息数量也是32,为此笔者猜测是Broker基于性能考虑，限制了一次拉取的消息数量最大是32。下面代码指定了一次批量拉取消息的最大数量是30,一次消费消息的最大数量是5。

@Test
public void testConsumeBatch() throws Exception 
  DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
  consumer.setNamesrvAddr(this.nameServer);
  //指定批量消费的最大值，默认是1
  consumer.setConsumeMessageBatchMaxSize(5);
  //批量拉取消息的数量，默认是32
  consumer.setPullBatchSize(30);
  consumer.subscribe("topic1", "tag6");
  consumer.registerMessageListener(new MessageListenerConcurrently() 
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) 
      System.out.println(Thread.currentThread().getName() + "一次收到" + msgs.size() + "消息");
      return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    
  );
  consumer.start();
  TimeUnit.SECONDS.sleep(120);
  consumer.shutdown();

当指定了批量消费消息的最大数量后，这一批消息将拥有同一个消费状态，即如果消息的消费状态返回的是CONSUME_SUCCESS，则它们都消费成功。如果返回的是RECONSUME_LATER，则它们都将再次投递。另外如果批量消费消息的数量不是1,则一次批量传递的消息都将在一个消费者线程中进行消费。比如指定了批量消费消息的最大数量是5,则如果拉取到了10条消息，则这10条消息会分别传递给两个消费者线程，每个线程各5条消息；而如果批量消费消息的数量使用默认值1则10条消息将传递给10个消费者线程，这样可能消费的性能更佳。读者朋友应该基于这两种情况慎重的考虑是否应该更改批量消费消息的默认值。

（注：本文是基于RocketMQ4.5.0所写）