企业官网建设流程全解析

学Kafka学到消费者这一块，我才真正意识到为什么网上那么多Kafka"重启就翻车"的案例——绝大多数最后都能归到offset这一件事上。今天这篇是Kafka系列的第三篇，接着前面生产者代码往下走，把消费者代码里的offset拆开揉碎了讲清楚。这篇内容不挑语言，虽然我代码示例用的是Linux下的C/C++和librdkafka，但offset这套概念在Java、Go里完全一样，理解之后换语言就是换个API的事。

1. 先把offset的基本概念盘明白：不是全局编号，是分区内的门牌号

1.1 offset到底是什么

一句话版本：offset是消息在分区内的顺序编号，从0开始递增，每一条消息在所属分区里有且只有一个唯一的offset，它决定了这条消息在分区里的先后位置。

我更喜欢用门牌号来理解它。一个Topic下面挂着好几个分区（Partition），每个分区就是一栋独立的楼，楼里的每个房间都有一个门牌号。消息进分区，就相当于住进了某个房间，门牌号就是offset。你要找一个楼里的所有住户，从头走到尾，顺序就是offset从小到大的顺序。

这里要特别强调一个新手容易踩的点：offset不是Topic级别的全局编号。同样是offset=100的消息，在分区0和分区1里是完全不同的两条消息，没有任何关联。你给消息做唯一标识的时候，必须用"Topic + Partition + Offset"三个值合起来才算数，少一个都不行。

另外，消息的offset一旦分配就不会再变。消息存多久，offset就保持多久，除非消息因为过期清理被删掉了，那这条offset就相当于从这个分区里"消失"了。这也是后面讲offset越界时会用到的前提。

1.2 position、committed offset和auto.offset.reset，三兄弟别搞混

真正开始写消费者代码后，我发现offset这个词在不同的语境下指的不是同一个东西。在Kafka的官方文档和librdkafka的头文件里，至少有三个相关概念：

1. position（当前读取位置）：消费者进程内存里维护的一个状态，表示"下一条要读的消息是哪个offset"。它跟着poll的进度一直往前走，进程一退出就没了。

2. committed offset（已提交的offset）：消费者把某个分区的读取进度同步到了Kafka服务端，存在一个叫__consumer_offsets的内部Topic里。这个值是持久化的，消费者重启之后，靠它恢复读取进度。

3. auto.offset.reset（重置策略）：当某个分区"没有可用已提交offset"或者"已提交offset越界"时，用这个配置决定从哪开始读。可选值是earliest、latest、none，默认是latest。

这三者的关系可以用一个场景串起来：你的消费者凌晨三点宕机了，重启之后，它先去__consumer_offsets里查自己这个消费组在某个分区上最后提交的offset是多少。如果查到了且这个值仍然在分区的有效范围内，就从那个offset接着读，跟没宕机一样。如果查不到（比如第一次启动，或者换了个全新的group.id），那就轮到auto.offset.reset出场了，earliest从头读、latest只读新消息、none直接报错。

这里有一个特别容易产生误解的名词——auto.offset.reset里的"reset"。它不是"随时重置消费进度"的意思，不是你想让它从头读它就能从头读的。只有当Kafka确认这个分区对你这个消费组"没有有效提交记录"时，reset才会生效。如果你想手动把消费进度退回去重新消费，靠的不是这个配置，而是后面要讲的seek或者重新提交旧offset。

还有一个很多人没注意到的细节：已提交offset的含义是"下一条待消费消息的offset"。比如你已经消费完了offset=99那条消息，提交的不是99，而是100。这跟普通数组遍历的"当前读到了哪个下标"的直觉是反的，很多同学写手动提交代码的时候在这里栽过跟头——提交了当前消息的offset，结果重启后那条消息又被读了一遍，造成重复消费。

2. 消费者代码里offset的三种来源：配置决定起点，自动/手动提交决定进度

2.1 auto.offset.reset的生效时机，比你想的更严格

我在学习日记里专门整理过一个表格，把reset的生效场景和结果对照起来，先贴出来：

这张表我建议收藏一下，尤其是第三行和第四行。也就是说，就算你已经跑了好几个月，只要__consumer_offsets里的提交记录因为某种原因失效了（最常见的就是消息超过retention.ms被清理了，提交的offset已经小于log start offset），reset策略依然会生效。这个"静默生效"很容易让人以为Kafka出bug了。

2.2 enable.auto.commit，方便但坑也不少

Kafka消费者默认开了自动提交，也就是enable.auto.commit=true。librdkafka里对应的配置项名一样。它的工作方式是这样的：消费者每poll一批消息，内部就记录一下当前读到了哪个offset，然后每隔auto.commit.interval.ms（默认5000毫秒）异步地把进度提交一次。

为什么说它"方便但坑不少"？因为它提交的是"最近一次poll到的offset"，而不是"业务处理完成后的offset"。假如你的消费流程是：

1. poll拉到一批消息
2. 写入本地数据库
3. 更新缓存

而第二步和第三步比较慢，超过5秒钟还没做完，这时候自动提交已经把offset往前推进了。万一第三步之后、处理完之前进程崩了，重启之后，Kafka告诉你"你已经消费到offset=200了"，但实际上offset=201到205的消息根本没有处理——这批消息就静默丢了。

所以在"处理成功才能算消费成功"的业务里，千万别用自动提交。这不是说自动提交一无是处，如果你只是做日志转发、写个统计脚本，消息丢了也无所谓，或者你的消费逻辑很简单不会超过提交间隔，那自动提交确实省心。

2.3 什么时候必须手动提交

我给自己定了三条标准，只要命中一条就走手动提交：

• 消费过程中有外部IO（写库、调接口、写文件）
• 要求at-least-once（至少一次）语义，宁可重复消费也不要丢消息
• 消费速度不均匀，处理时长可能波动很大

手动提交之后，commit本身也有同步和异步之分。librdkafka的rd_kafka_commit_message最后一个参数是同步标志，0表示同步，1表示异步。同步提交的好处是commit返回时你确定Kafka收到了，坏处是阻塞当前线程，吞吐量会下降；异步提交不阻塞，但你无法立即判断是否失败。我的习惯是：单分区消费用同步提交保证稳妥，多分区高吞吐场景用异步提交再加个回调检测失败情况。

3. 用librdkafka实操offset：订阅、拉取、手动提交全流程

3.1 一个能跑的消费者骨架

先看一段完整的初始化代码。我用的librdkafka版本是1.9.x，API在不同版本间略有差异，但主体逻辑稳定。

#include <librdkafka/rdkafka.h> #include <stdio.h> #include <signal.h> #include <inttypes.h> #include <string.h> static volatile sig_atomic_t g_run = 1; static void sig_handler(int sig) { g_run = 0; } int main(void) { char errstr[512]; rd_kafka_conf_t *conf = rd_kafka_conf_new(); /* 基础连接配置 */ rd_kafka_conf_set(conf, "bootstrap.servers", "127.0.0.1:9092", errstr, sizeof(errstr)); rd_kafka_conf_set(conf, "group.id", "study-group-01", errstr, sizeof(errstr)); /* 没有有效提交记录时从头消费 */ rd_kafka_conf_set(conf, "auto.offset.reset", "earliest", errstr, sizeof(errstr)); /* 今天先关掉自动提交，专注手动提交流程 */ rd_kafka_conf_set(conf, "enable.auto.commit", "false", errstr, sizeof(errstr)); rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr)); if (!rk) { fprintf(stderr, "create consumer failed: %s\n", errstr); return 1; } /* 创建高级消费者实例 */ rd_kafka_consumer_t *rk_c = rd_kafka_consumer_new(rk, "study-group-01", NULL); /* 订阅Topic */ rd_kafka_topic_partition_list_t *topics = rd_kafka_topic_partition_list_new(1); rd_kafka_topic_partition_list_add(topics, "study-topic", RD_KAFKA_PARTITION_UA); rd_kafka_resp_err_t err = rd_kafka_subscribe(rk_c, topics); if (err) { fprintf(stderr, "subscribe failed: %s\n", rd_kafka_err2str(err)); return 1; } signal(SIGINT, sig_handler); while (g_run) { rd_kafka_message_t *rkm = rd_kafka_consumer_poll(rk_c, 1000); if (!rkm) { continue; } if (rkm->err) { if (rkm->err != RD_KAFKA_RESP_ERR__PARTITION_EOF) { fprintf(stderr, "consumer error: %s\n", rd_kafka_message_errstr(rkm)); } } else { printf("partition=%"PRId32" offset=%"PRId64" msg=%.*s\n", rkm->partition, rkm->offset, (int)rkm->len, (char *)rkm->payload); } rd_kafka_message_destroy(rkm); } rd_kafka_consumer_close(rk_c); rd_kafka_destroy(rk); return 0; }

编译时记得链接-lrdkafka，运行前确认broker地址和Topic名正确。这段代码最重要的是看rkm->offset这个字段——它是librdkafka在rd_kafka_message_t结构体里直接暴露的消息元数据，你不用额外去查，poll回来的每条消息都自带offset。

这里再补一个新手常常忽略的细节：rd_kafka_consumer_poll返回的rkm，用完必须调用rd_kafka_message_destroy释放。消息的payload内存是随消息一起管理的，不释放会在长时间运行的程序里造成内存持续上涨，跟offset无关但会干扰你排查问题。

3.2 手动提交：commit message和commit list两种姿势

把上面代码的消费分支改成手动提交，核心逻辑是拿到消息、处理业务、然后提交。注意提交的时机一定是在业务处理完成之后，不能一拿到消息就提交。

/* 拿到消息（省略poll代码） */ rd_kafka_message_t *rkm = rd_kafka_consumer_poll(rk_c, 1000); if (!rkm || rkm->err) { /* 错误消息不需要提交 */ if (rkm) rd_kafka_message_destroy(rkm); continue; } /* 1. 处理业务：这里是模拟，实际可能是写库、调接口等 */ process_message(rkm->payload, rkm->len); /* 2. 处理成功之后，同步提交这条消息对应的offset */ rd_kafka_resp_err_t commit_err = rd_kafka_commit_message(rk_c, rkm, 0 /* 同步 */); if (commit_err) { fprintf(stderr, "commit message failed: %s\n", rd_kafka_err2str(commit_err)); /* 这里可以决定是重试还是记录待补偿 */ } rd_kafka_message_destroy(rkm);

这里要重点解释rd_kafka_commit_message的offset语义。这个API是"提交消息的offset"，但内部实现自动帮你做了加一操作，即提交的是这条消息offset加一的结果。前面讲过，已提交offset表示"下一条要消费的消息从哪开始"，所以提交当前消息N，Kafka记录的是N+1。我当初第一次看这个API的文档，一度以为它提交的是N，后来调试发现重复消费才弄明白，写代码时千万别自己再做一次加一，那就变成N+2提交了，会直接跳消息。

如果你的业务是批量处理的，一次poll拉了100条，全部处理成功后，可以用提交列表的方式一次提交完，减少commit的次数：

/* 自己构造一个offset列表，提交"本批次最后一条消息的offset + 1" */ rd_kafka_topic_partition_list_t *offsets = rd_kafka_topic_partition_list_new(1); rd_kafka_topic_partition_t *tpar = rd_kafka_topic_partition_new("study-topic", rkm->partition); tpar->offset = last_offset_of_batch + 1; rd_kafka_topic_partition_list_add(offsets, tpar); rd_kafka_resp_err_t batch_commit_err = rd_kafka_commit(rk_c, offsets, 0); if (batch_commit_err) { fprintf(stderr, "batch commit failed: %s\n", rd_kafka_err2str(batch_commit_err)); } rd_kafka_topic_partition_list_destroy(offsets);

这种批量提交方式有个前提：你确定这一批消息里的每一条都处理成功了。如果最后一条成功但中间某条失败了，这一批都得重来，靠下游做幂等去重兜底。批量构建rd_kafka_topic_partition_list_t的时候要小心，一个分区只能有一个条目，同一个分区追加多个条目会覆盖还是报错取决于版本，我习惯先查一遍再构造。

3.3 一个容易忽略的中间态：offsets_store只存不提交

librdkafka里还有一个容易被忽略的API叫rd_kafka_offsets_store。它的作用是把你处理过的offset存到消费者本地内存里，但不提交到broker。这个API配合enable.auto.commit=true使用时，自动提交的就不再是"最近poll到的offset"，而是你通过offsets_store显式确认过的offset。

这套组合拳能解决一个经典问题：既想要自动提交的省心，又想保证"处理成功才算数"。你关闭enable.auto.commit后的自动提交能力，改为处理成功后手动调rd_kafka_offsets_store，然后在合适的时机触发提交。不过说实话，这套机制首次接触时理解成本不低，我建议先把纯手动提交跑熟，再来看offsets_store，否则容易把自动提交和手动提交的逻辑混在一起，排错时非常痛苦。

4. seek和查询offset的API：消费进度由你掌控

4.1 seek：把消费指针拨回去

手动提交解决的是"进度往前进"的问题，那如果我想让消费进度倒退，重新消费一批数据呢？这时候用rd_kafka_seek。

/* 把分区0的消费位置重置到offset=500 */ rd_kafka_topic_partition_t *tpar = rd_kafka_topic_partition_new("study-topic", 0); tpar->offset = 500; rd_kafka_resp_err_t seek_err = rd_kafka_seek(rk_c, tpar, 0); if (seek_err) { fprintf(stderr, "seek failed: %s\n", rd_kafka_err2str(seek_err)); } rd_kafka_topic_partition_destroy(tpar);

seek的offset语义和提交是一样的：把位置拨到500，下一条读的就是offset=500的消息。如果你希望"读完500以后不要了"，就拨到501。

seek有几个限制条件，我踩过之后印象很深：

• 只能seek当前已经分配给这个消费者的分区。消费者没订阅这个Topic，或者分区还在重平衡（rebalance）过程中，调用会返回RD_KAFKA_RESP_ERR__STATE错误。
• seek不会自动更新committed offset。它只改内存里的读取位置，如果你seek完又什么都不提交就退出，下次重启还是会从老的提交位置读，你的seek等于白做。想要"重放后持久生效"得配合手动提交。
• 订阅模式下对正在poll的消费实例做seek会有竞态问题。librdkafka的poll线程和你的主线程是并发的，seek和poll同时进行时，可能出现seek完了但poll还返回了旧位置的消息。稳妥做法是暂停（pause）这个分区，seek，再恢复（resume）。

最后一个限制很多文章没写，但实际项目里重放数据时特别容易触发。我现在处理重放需求的顺序是：先调rd_kafka_pause_partitions把分区暂停，再seek，最后rd_kafka_resume_partitions恢复。这样能确保seek之后poll到的消息一定是从新位置开始的。

4.2 query_watermark_offsets：看分区的水位线

除了seek，日常运维还有一个高频需求：看这个分区现在有多少消息了，消费到哪了，积压了多少。

int64_t low = 0; int64_t high = 0; rd_kafka_resp_err_t err = rd_kafka_query_watermark_offsets( rk, "study-topic", 0, &low, &high, 5000); if (!err) { printf("partition 0: log_start_offset=%"PRId64 " log_end_offset=%"PRId64"\n", low, high); }

这里的low就是log start offset，也就是分区里现存最早的消息offset；high是log end offset，是"下一条要写入的消息的offset"（不是最后一条消息的offset）。所以分区的消息总量不是high - low + 1，而是high - low，这个细节也容易算错。

配合消费者的当前position，你就能算出消费积压（lag）：lag = high - consumer_position。consumer_position可以通过rd_kafka_position拿到当前各分区的position。这个组合是监控消费进度是否跟得上的基础工具，我自己的脚本里就是定时拉这两个值，超过阈值就告警。

4.3 offsets_for_times：按时间戳找offset

还有一类场景：我只想重新消费"从昨天下午3点开始"的数据，但不知道那时的offset是多少。手动去翻日志找offset不现实，Kafka提供了一个按时间戳查offset的API，rd_kafka_offsets_for_times。

/* 目标时间戳，单位毫秒 */ int64_t target_timestamp = 1700000000000LL; rd_kafka_topic_partition_list_t *parts = rd_kafka_topic_partition_list_new(1); rd_kafka_topic_partition_t *tpar = rd_kafka_topic_partition_new("study-topic", 0); tpar->offset = target_timestamp; /* 传入时是时间戳 */ rd_kafka_topic_partition_list_add(parts, tpar); rd_kafka_resp_err_t err = rd_kafka_offsets_for_times(rk, parts, 5000); if (!err) { /* 返回后，tpar->offset就是查到的消息offset */ printf("partition %d first offset >= timestamp: %"PRId64"\n", tpar->partition, tpar->offset); } rd_kafka_topic_partition_list_destroy(parts);

这个API的语义是：对每个分区，返回"时间戳大于等于你给定时间戳的第一条消息的offset"。原理是Kafka存储消息时，在索引里记录了按消息时间戳分组的位置信息，broker收到查询请求后直接在索引里二分查找。

用这个API有几个前提要知道：

• broker版本不能太老，时间戳索引是在0.10.1版本引入的，老版本broker不支持。
• 消息写入时如果没有指定时间戳，broker会用当前时间填充，所以时间索引基本都有值。
• 如果查询的时间戳早于分区里最早消息的时间戳，返回的offset就是log start offset；如果晚于最后一条消息的时间戳，返回的offset可能是-1，表示这个分区里没有符合条件的消息，代码里一定要对这个-1做判断，不能直接拿去seek。

5. 实际项目里踩过的offset相关的坑，每一个都值得你记住

5.1 改了group.id，以为会从指定位置读，结果从头开始

这是我第一次做Kafka消费重放时的真实教训。当时想把线上数据重新读一遍，想着改个group.id、把auto.offset.reset设成earliest，然后重启消费者。逻辑上觉得"新group没有提交记录，earliest就是从最早读"，确实，数据是从头读了。但问题在于：我用这个新group消费完之后，没有清理提交记录，也没删除group。第二次再用同一个group名跑，它以为自己有提交记录了，直接从上次的位置继续读，结果我期待的是从头读，实际却从中间开始。

从那以后我记住一条规则：group.id和消费语义强绑定。如果你只是想临时改一下消费逻辑跑一遍全量数据，跑完记得手动把group的offset清掉或者用别的工具删除group，否则这个group残留的提交记录会一直影响之后的运行。

5.2 手动提交放在处理前，一崩就丢数据

有段时间我维护的一个消费者程序偶尔会丢消息，查了很久最后定位到提交代码的位置不对。当时的代码大概是这样的：

rd_kafka_message_t *rkm = rd_kafka_consumer_poll(rk_c, 1000); rd_kafka_commit_message(rk_c, rkm, 0); /* 先提交 */ process_message(rkm->payload, rkm->len); /* 后处理 */

提交放在处理前，等于告诉Kafka"这条消息我已经搞定了"，但实际业务还没执行。一旦处理过程中进程崩溃或者机器宕机，消息就永久丢失。这个坑的根本原因是"先提交后处理"违背了at-least-once语义的基本要求——只有处理成功才能提交。这个顺序问题比什么API用错都隐蔽，因为它大多数时候都能正常工作，只在崩溃那一刻暴露问题。

5.3 分区重平衡丢消息的表象，本质还是offset没提交

另外一次排障经历是：消费者程序稳定运行，但每隔一段时间就有几条消息重复消费，把日志打开看，重复的消息都出现在同一段时间内，而且位置正好跨分区。最后确认是Consumer Rebalance引起的——消费者组成员变化（比如某个实例挂了或扩容了）时，分区会在成员之间重新分配，如果你的进度没有及时提交，接管分区的那个消费者会从旧提交位置重新拉取，于是出现"看起来卡顿一下，然后一批消息重新处理"的现象。

这种重复无法完全避免，分布式系统里at-least-once语义就是允许重复的，治理手段是让下游消费逻辑幂等，或者用事务性消息方案。我的经验是：先接受重复，再想办法去重，而不是跟分区重平衡机制较劲。

5.4 时间戳查offset时，多分区粒度不一致

用offsets_for_times做按时间重放时，还有一个容易忽视的细节：它是按分区分别查询的。假设你的Topic有4个分区，消息生产时间有先有后，同一个时间戳在不同分区的对应offset可能差别很大。你把4个分区都查一遍，拿到的4个offset分别seek回去，消费出来的数据并不严格等于"某个时间点之后的消息全集"，而是"每个分区各自在该时间点之后的消息"。

简单说，Kafka的消息没有全局时间序，只有分区内的时间序。如果你的业务要求严格按时间顺序重放，得先想清楚是单分区消费，还是接受分区间的轻微乱序，不能想当然地以为4个分区的查询结果拼起来就是一个整齐的时间轴。

6. 一点自己的学习经验

把offset这套概念和API捋顺之后，回头看Kafka消费者代码，其实核心就是两件事：拉消息的正确姿势和进度提交的正确时机。前者靠poll，后者靠你对offset语义的理解。我建议新手在本地起一个单节点Kafka，创建一个小Topic，手写一个消费者，故意制造几种异常场景——不提交重启、提交位置往前调、seek到越界offset，亲眼看看消息是怎么重复、怎么丢失、怎么报错的。这些现象看一遍比看十篇文档都管用。

还有一个小技巧分享给做C/C++的同学：调试offset相关的逻辑时，可以在环境变量里设置RDKAFKA_DEBUG=commit,consumer,cgrp再启动程序，librdkafka会把消费组状态、提交请求、每次提交的offset都打到标准错误输出里。当年定位手动提交加一那个问题，就是靠这个调试输出一锤定音的，比自己写日志省事太多。