大家好，我是君哥。
最近有读者参加面试时被问了一个问题，如果消费者拉取了一批消息，比如 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 类图如下：
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 免费无限量使用 DeepSeek R1 模型☜☜☜
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 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，这样必须考虑冥等问题。
