面试题：消息队列RocketMQ的消费者API如何处理消息积压并保证消息不丢失？

Question

当RocketMQ消费者面临消息积压问题时，如何利用消费者API进行有效的处理，既要保证处理效率，又要确保消息不会丢失？阐述在不同消费模式（集群消费、广播消费）下的处理策略及API使用要点。

Answer 1

集群消费模式

1. 增加消费者实例：
   • 处理策略：通过增加消费者实例数量，利用多实例并行消费消息，提高消费效率。RocketMQ会将队列均匀分配给不同的消费者实例。
   • API使用要点：在创建消费者实例时，配置合适的消费者组。例如，使用Java客户端，代码如下：

   DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("your_consumer_group");
   consumer.setNamesrvAddr("namesrv_ip:port");
   consumer.subscribe("your_topic", "*");
   consumer.registerMessageListener(new MessageListenerConcurrently() {
       @Override
       public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
           // 处理消息逻辑
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       }
   });
   consumer.start();
   

2. 提高单个消费者消费线程数：
   • 处理策略：对于单个消费者实例，增加其内部的消费线程数，加快消息处理速度。
   • API使用要点：以Java客户端为例，可以通过DefaultMQPushConsumer的setConsumeThreadMin和setConsumeThreadMax方法设置消费线程的最小和最大数量。

   DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("your_consumer_group");
   consumer.setConsumeThreadMin(20);
   consumer.setConsumeThreadMax(50);
   

3. 批量消费：
   • 处理策略：消费者一次拉取并处理多条消息，减少拉取消息的网络开销，提高消费效率。
   • API使用要点：在Java客户端中，通过DefaultMQPushConsumer的setConsumeMessageBatchMaxSize方法设置每次消费的最大消息数量。

   DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("your_consumer_group");
   consumer.setConsumeMessageBatchMaxSize(10);
   

4. 消息重试：
   • 处理策略：当消息消费失败时，RocketMQ会自动进行重试，确保消息不会丢失。
   • API使用要点：消费失败时，返回相应的消费状态，如ConsumeConcurrentlyStatus.RECONSUME_LATER 。

   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
       try {
           // 处理消息逻辑
           return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       } catch (Exception e) {
           return ConsumeConcurrentlyStatus.RECONSUME_LATER;
       }
   }
   

广播消费模式

1. 提高单个消费者消费线程数：
   • 处理策略：与集群消费类似，通过增加单个消费者实例内部的消费线程数来加快消息处理速度。由于广播消费每个消费者实例都会收到所有消息，提高消费线程数有助于提升处理效率。
   • API使用要点：同样在Java客户端中，通过DefaultMQPushConsumer的setConsumeThreadMin和setConsumeThreadMax方法设置消费线程的最小和最大数量。

   DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("your_consumer_group");
   consumer.setConsumeThreadMin(20);
   consumer.setConsumeThreadMax(50);
   

2. 批量消费：
   • 处理策略：和集群消费一样，广播消费也可以采用批量消费的方式，一次拉取并处理多条消息，减少网络开销，提高消费效率。
   • API使用要点：通过DefaultMQPushConsumer的setConsumeMessageBatchMaxSize方法设置每次消费的最大消息数量。

   DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("your_consumer_group");
   consumer.setConsumeMessageBatchMaxSize(10);
   

3. 消息重试：
   • 处理策略：广播消费模式下，消息消费失败不会自动重试。需要在业务代码中自行实现重试逻辑，确保消息不丢失。
   • API使用要点：消费失败时，在业务代码中进行重试。例如：

   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
       int retryCount = 3;
       for (int i = 0; i < retryCount; i++) {
           try {
               // 处理消息逻辑
               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
           } catch (Exception e) {
               if (i == retryCount - 1) {
                   // 记录日志等处理
                   return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
               }
           }
       }
       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
   }