大家好，我是君哥。
最近有读者参加面试时被问了一个问题，如果消费者拉取了一批消息，比如 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() {@Overridepublic 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 类图如下：
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 免费无限量使用 DeepSeek R1 模型☜☜☜
Typewise.app
面向客户服务和销售团队的AI写作解决方案。
39 查看详情
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 = 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() {@Overridepublic 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，这样必须考虑冥等问题。
以上就是阿里二面：RocketMQ 消费者拉取一批消息，其中部分消费失败了，偏移量怎样更新？的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/838453.html
微信扫一扫 
支付宝扫一扫 
