面试题：RocketMQ 的消费位点（Offset）是如何管理的？

RocketMQ 的消费位点管理是其消息可靠传递的核心机制，主要通过以下几种方式实现：

1. Offset 存储位置

1.1 本地模式 (默认)

• 存储路径：~/.rocketmq_offsets/{consumerGroup}/offsets.json
• 特点：
  • 消费者本地维护
  • 简单高效
  • 消费者重启后可能重复消费

1.2 Broker 模式 (集群模式)

• 存储位置：Broker 服务器的 {rocketmq.home}/store/config/consumerOffset.json
• 特点：
  • 集中式管理
  • 支持消费者动态扩容
  • 消费进度不会丢失

2. Offset 提交机制

2.1 自动提交 (默认)

// 配置自动提交间隔（默认5秒）
consumer.setPersistConsumerOffsetInterval(5000);

2.2 手动提交

// 在消息监听器中手动提交
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
        ConsumeConcurrentlyContext context) {
        // 处理消息...
        context.setAckIndex(msgs.size() - 1); // 设置确认位置
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

3. Offset 核心管理流程

1. 初始化阶段：
   • 消费者启动时从存储位置读取上次的消费位点
   • 如果首次启动，根据配置决定从最早或最新开始消费
2. 运行时更新：
   • 成功消费一批消息后更新内存中的位点
   • 定时/定量触发位点持久化
3. 异常处理：
   • 消费失败时位点不会前进
   • 支持重试队列机制

4. 重要配置参数

# 从何处开始消费 (CONSUME_FROM_LAST_OFFSET/CONSUME_FROM_FIRST_OFFSET)
consumeFromWhere=CONSUME_FROM_LAST_OFFSET

# 位点持久化间隔 (ms)
persistConsumerOffsetInterval=5000

# 位点同步/异步提交
offsetStore.sync=true

5. 特殊场景处理

5.1 重置消费位点

// 从指定时间开始消费
consumer.resetOffsetByTimeStamp(topic, new Date().getTime() - 3600*1000);

5.2 顺序消费位点

• 顺序消费采用队列锁机制
• 位点提交必须严格有序

5.3 广播模式

• 每个消费者实例独立维护自己的位点
• 不进行位点同步

6. 监控与管理

1. 控制台查看：sh mqadmin consumerProgress -n namesrv:9876 -g consumerGroup
2. 关键指标：
   • brokerOffset：Broker最新消息位点
   • consumerOffset：消费者已消费位点
   • diff：积压消息量
3. 运维命令：sh mqadmin resetOffsetByTime -n namesrv:9876 -g group -t topic -s 1

RocketMQ 的位点管理机制既保证了消息的可靠传递，又提供了灵活的配置选项，开发者需要根据业务场景选择合适的位点管理策略。