注：该文用于个人学习记录和知识交流，如有不足，欢迎指点。

一、offset是什么？

消息 offset：Kafka 分区内消息的唯一递增序号，标识消息位置；

消费 offset：消费者提交的位移，代表下一次要消费的消息序号。

例子：

• A 消费了 offset 800、801、802（最新接收的是 802）；
• A 自动提交了 offset 803（客户端默认提交已拉取的最大 offset + 1）；
• B 接手后，直接消费 offset 803 的消息（即 802 + 1）；
• 若 A 只提交了 801（非最新），B 则从 801 开始消费，会重复消费 801、802，再消费 803 及以后

简单点讲：提交 N，消费 N；已消费到 N-1，提交 N。

二、消费者offset的相关api

1. offset的常量

常量	数值	含义
OFFSET_BEGINNING	-2	从分区第一条消息开始消费
OFFSET_END	-1	从分区最新消息的下一条开始消费
OFFSET_STORED	-1000	使用消费组存储的位移

注意：“分区最开始（第一条消息）” 指的是当前磁盘上实际存在的最早消息，而不是固定的 offset 0。

核心原因：日志清理会改变分区的起始偏移量

例子：

• 原始分区有 100000 条消息，offset 从 0 到 99999；
• 由于日志保留策略，offset 0-49999 的日志段被删除；
• 此时，分区中最早的消息是 offset 50000；
• 当你使用 OFFSET_BEGINNING 时，消费者会从 offset 50000 开始消费，而不是 0。

2. 客户端offset的自动提交

在 librdkafka C++ 客户端 中，这两个配置项有明确的默认值：

配置项	默认值	含义
enable.auto.commit	true	默认开启自动提交 offset
auto.commit.interval.ms	5000	默认每 5 秒自动提交一次

也就是说，即使你没有在代码中显式调用 set 方法，这两个配置也会生效，你的消费者会默认以 5 秒为周期自动提交 offset。

2.1 自动提交的默认行为

• 周期性提交：每 5 秒，客户端会自动异步提交当前所有分区的最大已拉取 offset。
• 重平衡触发提交：当消费组发生重平衡（如消费者加入 / 退出）时，客户端会同步提交一次 offset，避免重平衡后重复消费。
• 消费者关闭提交：调用 m_consumer->close() 时，会触发一次同步提交，确保最后消费的 offset 被保存。

2.2 生产环境的关键提醒

虽然自动提交很方便，但它存在一个核心风险：只保证消息被拉取，不保证业务处理成功。

• 如果消费者拉取了消息（offset 1000），但在业务处理（如入库）完成前崩溃，而自动提交已经提交了 offset 1000，那么这条消息就会丢失。
• 因此，在生产环境中，强烈建议关闭自动提交，改为在业务处理成功后手动提交 offset：

  // 关闭自动提交
  m_config->set("enable.auto.commit", "false", errorStr);
  

3.  客户端有关offset的相关api

3.1 核心配置项（offset 相关）

配置项	用途	默认值
enable.auto.commit	开关自动提交 offset	true
auto.commit.interval.ms	自动提交间隔（ms）	5000
auto.offset.reset	无有效 offset 时的兜底策略（earliest/latest/none）	latest

3.2 核心 API（C++，librdkafka）

API	用途
commitSync()/commitSync(partitions)	消费者同步提交 offset（阻塞，保证落地）
commitAsync()/commitAsync(partitions)	消费者异步提交 offset（非阻塞，高性能）
position(partitions)	消费者获取当前消费位置（未提交的 offset）
committed(partitions, timeout)	消费者获取已提交的 offset（消费组存储的位移）
seek(tp, offset, timeout)	消费者手动定位到指定 offset 消费
assignment()	消费者获取当前分配的分区列表（用于精准提交）
RdKafka::TopicPartition::set_offset(offset)	为分区对象设置目标消费 offset（搭配 assign/seek 使用）

4. commitAsync、commitSync

4.1 参数

调用方式	是否可手动指定 offset	提交的 offset 来源
commitAsync()/commitSync()	否	客户端自动取「每个分区已消费的最大 offset + 1」作为提交值（即下一次要消费的 offset）
commitAsync(partitions)/commitSync(partitions)	是	提交 partitions 列表中每个 TopicPartition 通过 set_offset() 指定的 offset

示例（手动指定 offset 提交）：

// 构造要提交的分区+offset
std::vector<RdKafka::TopicPartition*> partitions;
RdKafka::TopicPartition* tp = RdKafka::TopicPartition::create("test", 0);
tp->set_offset(1000); // 手动指定提交 offset=1000
partitions.push_back(tp);

// 同步提交指定的 offset
m_consumer->commitSync(partitions);

// 释放资源
for (auto p : partitions) delete p;

日常批量提交用无参数版本，无需手动指定，客户端自动处理；

精准控制（如重置位移、补提位移）用带参数版本，需通过 set_offset() 手动指定

4.2 需要手动set_offset的情况

提交方式	是否需要手动 set_offset	offset 来源	适用场景
commitSync(partitions)（重平衡回调中）	❌ 不需要	客户端自动填充的当前消费位置	重平衡时提交最新位移，保证一致性
commitSync(partitions)（用户手动构造）	✅ 需要	用户通过 set_offset() 指定	重置位移、补提历史位移等特殊操作

订阅前手动指定从哪里开始消费（例子）：

    // ========== 关键改动1：构造指定的「主题+分区」列表 ==========
    std::vector<RdKafka::TopicPartition*> assign_partitions;
    // 遍历要消费的主题列表，为每个主题指定要消费的分区（示例：使用成员变量m_partition，也可自定义多个分区）
    for (const auto& topic : m_topicVector) {
        // 创建「主题+分区」对象（核心：指定具体分区号，比如m_partition=0）
        RdKafka::TopicPartition* tp = RdKafka::TopicPartition::create(topic, m_partition);
        // 【可选】指定消费起始位移（比如从第一条消息开始消费，不指定则用auto.offset.reset）
        tp->set_offset(RdKafka::Topic::OFFSET_STORED); 
        /**
         * RdKafka::Topic::OFFSET_BEGINNING：从第一条消息开始消费 (注意值为-2，并非为0)
         * RdKafka::Topic::OFFSET_STORED:使用消费组存储的历史位移， （注意值为-1000）
         * RdKafka::Topic::OFFSET_END：只消费未来新消息 （注意值为-1）
         */

    
        
        assign_partitions.push_back(tp);
    }

    // ========== 关键改动2：用assign替换subscribe，手动分配分区 ==========
    RdKafka::ErrorCode errorCode = m_consumer->assign(assign_partitions);

4.3 提交offset常见的报错

Local: No offset stored

这个错误：表示位移已经重复提交，或者说根本就没有消费过。