阿里二面：RocketMQ 消费者拉取一批消息，其中部分消费失败了，偏移量怎样更新？

大家好，我是君哥。

最近有读者参加面试时被问了一个问题，如果消费者拉取了一批消息，比如 100 条，第 100 条消息消费成功了，但是第 50 条消费失败，偏移量会怎样更新？就着这个问题，今天来聊一下，如果一批消息有消费失败的情况时，偏移量怎么保存。

1 拉取消息

1.1 封装拉取请求

以 RocketMQ 推模式为例，RocketMQ 消费者启动代码如下：

public static void main(String[] args) throws InterruptedException, MQClientException {
 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");

 consumer.subscribe("TopicTest", "*");
 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

 consumer.setConsumeTimestamp("20181109221800");
 consumer.registerMessageListener(new MessageListenerConcurrently() {

  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){
   try{
    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
   }catch (Exception e){
    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
   }
   return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  }
 });
 consumer.start();
}

上面的 DefaultMQPushConsumer 是一个推模式的消费者，启动方法是 start。消费者启动后会触发重平衡线程（RebalanceService），这个线程的任务是在死循环中不停地进行重平衡，最终封装拉取消息的请求到 pullRequestQueue。这个过程涉及到的 UML 类图如下：

1.2 处理拉取请求

封装好拉取消息的请求 PullRequest 后，RocketMQ 就会不停地从 pullRequestQueue 获取消息拉取请求进行处理。UML 类图如下：

拉取消息的入口方法是一个死循环，代码如下：

//PullMessageService
public void run(){
 log.info(this.getServiceName() + " service started");

 while (!this.isStopped()) {
  try {
   PullRequest pullRequest = this.pullRequestQueue.take();
   this.pullMessage(pullRequest);
  } catch (InterruptedException ignored) {
  } catch (Exception e) {
   log.error("Pull Message Service Run Method exception", e);
  }
 }

 log.info(this.getServiceName() + " service end");
}

这里拉取到消息后，提交给 PullCallback 这个回调函数进行处理。

拉取到的消息首先被 put 到 ProcessQueue 中的 msgTreeMap 上，然后被封装到 ConsumeRequest 这个线程类来处理。把代码精简后，ConsumeRequest 处理逻辑如下：

//ConsumeMessageConcurrentlyService.java
public void run(){
 MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
 ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(messageQueue);
 ConsumeConcurrentlyStatus status = null;
 try {
  //1.执行消费逻辑，这里的逻辑是在文章开头的代码中定义的
  status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
 } catch (Throwable e) {
 }
 if (!processQueue.isDropped()) {
  //2.处理消费结果
  ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
 } else {
  log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);
 }
}

2 处理消费结果

2.1 并发消息

并发消息处理消费结果的代码做精简后如下：

//ConsumeMessageConcurrentlyService.java
public void processConsumeResult(
 final ConsumeConcurrentlyStatus status,
 final ConsumeConcurrentlyContext context,
 final ConsumeRequest consumeRequest
){
 int ackIndex = context.getAckIndex();
 switch (status) {
  case CONSUME_SUCCESS:
   if (ackIndex >= consumeRequest.getMsgs().size()) {
    ackIndex = consumeRequest.getMsgs().size() - 1;
   }
   int ok = ackIndex + 1;
   int failed = consumeRequest.getMsgs().size() - ok;
   break;
  case RECONSUME_LATER:
   break;
  default:
   break;
 }

 switch (this.defaultMQPushConsumer.getMessageModel()) {
  case BROADCASTING:
   for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
   }
   break;
  case CLUSTERING:
   List<MessageExt> msgBackFailed = new ArrayList<MessageExt>(consumeRequest.getMsgs().size());
   for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
    MessageExt msg = consumeRequest.getMsgs().get(i);
    boolean result = this.sendMessageBack(msg, context);
    if (!result) {
     msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
     msgBackFailed.add(msg);
    }
   }

   if (!msgBackFailed.isEmpty()) {
    consumeRequest.getMsgs().removeAll(msgBackFailed);
   }
   break;
  default:
   break;
 }

 long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());
 if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
  this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);
 }
}

从上面的代码可以看出，如果处理消息的逻辑是串行的，比如文章开头的代码使用 for 循环来处理消息，那如果在某一条消息处理失败了，直接退出循环，给 ConsumeConcurrentlyContext 的 ackIndex 变量赋值为消息列表中失败消息的位置，这样这条失败消息后面的消息就不再处理了，发送给 Broker 等待重新拉取。代码如下：

public static void main(String[] args) throws InterruptedException, MQClientException {
 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");

 consumer.subscribe("TopicTest", "*");
 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

 consumer.setConsumeTimestamp("20181109221800");
 consumer.registerMessageListener(new MessageListenerConcurrently() {

  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){
   for (int i = 0; i < msgs.size(); i++) {
    try{
     System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
    }catch (Exception e){
     context.setAckIndex(i);
     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
   }
   return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
 });
 consumer.start();
}

消费成功的消息则从 ProcessQueue 中的 msgTreeMap 中移除，并且返回 msgTreeMap 中最小的偏移量（firstKey）去更新。注意：集群模式偏移量保存在 Broker 端，更新偏移量需要发送消息到 Broker，而广播模式偏移量保存在 Consumer 端，只需要更新本地偏移量就可以。

如果处理消息的逻辑是并行的，处理消息失败后给 ackIndex 赋值是没有意义的，因为可能有多条消息失败，给 ackIndex 变量赋值并不准确。最好的方法就是给 ackIndex 赋值 0，整批消息全部重新消费，这样又可能带来冥等问题。

2.2 顺序消息

对于顺序消息，从 msgTreeMap 取出消息后，先要放到 consumingMsgOrderlyTreeMap 上面，更新偏移量时，是从 consumingMsgOrderlyTreeMap 上取最大的消息偏移量（lastKey）。

3 总结

回到开头的问题，如果一批消息按照顺序消费，是不可能出现第 100 条消息消费成功了，但第 50 条消费失败的情况，因为第 50 条消息失败的时候，应该退出循环，不再继续进行消费。

如果是并发消费，如果出现了这种情况，建议是整批消息全部重新消费，也就是给 ackIndex 赋值 0，这样必须考虑冥等问题。