大家好，我是君哥。

最近有读者参加面试时被问了一个问题，如果消费者拉取了一批消息，比如 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 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 类图如下：

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

//PullMessageServicepublic 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.javapublic 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.javapublic 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 msgBackFailed = new ArrayList(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 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，这样必须考虑冥等问题。

关键词： 消息处理 变量赋值 进行处理 发送消息