Golang使用sarama生产消费Kafka：核心配置与避坑指南

做Golang后端开发的同学基本都绕不开Kafka这块。标题里“golang使用sarama生产和消费kafka”看起来像是个入门操作但真要把生产者和消费者写得能扛住线上流量、不出幺蛾子里面还是有不少门道。sarama是Golang社区里使用最广的Kafka客户端库纯Go实现部署起来很干净但是它的参数多、行为细节也多我在项目里从最初“能发能收”到后来“稳定压测通过”踩了不少坑。这篇文章就针对sarama这套客户端把生产端和消费端从配置到代码到排查完整过一遍给正在接手或准备上Kafka的人一份能直接抄作业的参考。文章不会只贴两段能跑通的最小示例更多会聊配置为什么这么设、哪些参数是生产环境必须调的、哪些地方容易出诡异问题。适合刚从入门走向真实项目、或者已经被线上Kafka问题折磨过的开发者。1. 为什么选择sarama操作Kafka1.1 Golang的Kafka客户端生态盘点Golang里能用来操作Kafka的库主流的大概有三条路线。客户端底层实现优点缺点sarama纯Go无CGO依赖、部署简单、功能覆盖面广、社区资料多部分实现细节依赖版本、参数复杂、文档偏散kafka-go纯Go接口设计简洁、支持Reader/Writer语义、上手快高级特性支持相对弱一些遇到复杂场景扩展要自己来confluent-kafka-go包装librdkafkaC库性能天花板高、特性跟随Kafka版本很紧需要CGO交叉编译麻烦环境问题会比较折腾选sarama的原因很朴素第一它是纯Go放到容器里一个二进制就能跑不用在镜像里搞gcc那套第二它出来的时间早网上踩坑经验多真出问题查起来有路可走第三它同时覆盖生产者、消费者、消费组、AdminClient、Mock测试这些能力一个库基本把Kafka相关的活儿都包了。现在的sarama包路径是github.com/IBM/sarama早期是github.com/Shopify/sarama2021年前后迁移到IBM组织下继续维护。线上代码里两种引用路径都可能见到遇到老项目别觉得奇怪功能上基本延续。1.2 sarama的核心优势与代价sarama最实用的部分是对Kafka协议细节的封装非常完整。比如生产者端的分区选择、消息确认机制、批量发送、压缩消费者端的消费组协调、offset提交、重平衡处理这些高层概念都被抽象成了比较清晰的接口。写业务代码时不太需要关心底层协议帧的组装解析。但代价也很明显配置项多到让人头大。Producer结构体里面几十个字段Consumer结构体里几十个字段很多还有默认值。不同版本之间的默认行为有差异比如某个版本开始Consumer.Return.Errors默认值变了某个版本Producer.Idempotent的校验更严格了。这些细节点靠读文档是不够的很多时候得翻源码确认。所以我觉得sarama的学习思路应该是先用最小代码跑通流程建立起“消息怎么发、offset怎么管理”的感知再去逐个吃透配置项。直接对着文档一个个配容易被淹没在参数海洋里。这篇文章也是按这个逻辑来组织先把基础链路跑通再讲清楚核心参数。2. 环境准备与必备概念2.1 快速起一个本地Kafka环境本地开发和联调最省事的方式是用Docker跑一个单节点Kafka。我常用的docker-compose.yml简化一下大概是这样的version: 3 services: kafka: image: bitnami/kafka:3.5 ports: - 9092:9092 environment: - KAFKA_CFG_NODE_ID0 - KAFKA_CFG_PROCESS_ROLEScontroller,broker - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS0kafka:9093 - KAFKA_CFG_LISTENERSPLAINTEXT://:9092,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://localhost:9092 - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLEtrue注意最后一个环境变量AUTO_CREATE_TOPICS_ENABLEtrue开发环境建议开着不然消费者去消费一个不存在的topic时Kafka会自动创建还是拒绝要看broker配置某些版本下消费者会一直拿不到分区表现很诡异。生产环境一般建议关掉自动创建topic由运维统一管理这个区别要先记住。起好环境后可以先手动创建topic方便测试kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test-topic --partitions 3 --replication-factor 1分区数设成3就够模拟多分区消费的场景了。2.2 sarama里绕不开的几个核心概念在写代码前有几个概念如果没建立清楚后面看日志和调参数会非常痛苦。消息MessageKafka里传递的是一条条消息包含key、value、headers、timestamp等字段。key不为空时会参与分区路由。分区Partition一个topic可以拆成多个分区每个分区内部是有序的消息按offset递增排列。可以类比成一本书被拆成几册每册页码连续但不同册之间的顺序不保证。offset偏移量分区内每条消息的序号从0开始递增。消费者需要记录自己读到哪个offset了这就是“消费位点”。消费组Consumer Group一组消费者共同消费一个topic组内每个消费者负责部分分区。这是Kafka实现水平扩展和故障转移的基础机制。重平衡Rebalance消费组成员变化、分区数量变化时组内会重新分配分区与消费者的对应关系。重平衡期间消费会中断这也是后面要讲的重点坑。sarama里的ConsumerGroup接口就是围绕消费组这套机制设计的。如果只是单纯读数据不用消费组也可以用Consumer接口手动管理分区和offset但生产环境基本都走消费组。3. 生产者从发送到确认的完整实现3.1 生产者配置参数详解先看一段我常用的生产者初始化代码package main import ( time github.com/IBM/sarama ) func newProducer() (sarama.SyncProducer, error) { config : sarama.NewConfig() // 生产端必须显式指定版本避免协议兼容问题 config.Version sarama.V3_5_0_0 // 等待所有ISR都写入成功才返回成功 config.Producer.RequiredAcks sarama.WaitForAll // 发送失败时的重试次数配合幂等使用 config.Producer.Retry.Max 5 config.Producer.Retry.Backoff 100 * time.Millisecond // 允许单个ProduceRequest里包含多条消息提高吞吐 config.Producer.Flush.Frequency 10 * time.Millisecond config.Producer.Return.Successes true config.Producer.Return.Errors true // 打开幂等producer会保证消息不重复、不丢 config.Producer.Idempotent true // 单链接未确认请求的最大数量幂等模式下只允许1 config.Net.MaxOpenRequests 1 producer, err : sarama.NewSyncProducer([]string{localhost:9092}, config) if err ! nil { return nil, err } return producer, nil }重点解释几个关键参数Version这个必须显式设不要用sarama.MinVersion或者默认的旧版本。Kafka协议是向后兼容的但高版本特性比如新的消息格式、分组协调协议需要客户端的版本足够高。如果客户端版本比broker低某些特性会静默降级日志排查非常费劲。我一般直接跟broker主版本一致比如Kafka 3.5就用V3_5_0_0。RequiredAcks有三个档位NoResponse不等待、WaitForLocal等待leader写入、WaitForAll等待所有ISR写入。从语义上理解WaitForAll是最安全的配合MinISR配置可以保证消息写入到足够多的副本后才返回成功。代价是链路延迟会比WaitForLocal高一些但大部分业务系统值得这个延迟。Idempotent Retry.Max Net.MaxOpenRequests这是生产环境最容易忽略的一组参数组合。开启幂等后producer会为每条消息带上序列号broker端对重复序列号做去重这样即使重试也能保证不产生重复消息。而Kafka协议要求幂等生产者必须保证分区内消息的顺序所以同一分区同时只能有一个未确认请求MaxOpenRequests必须为1否则启动producer会直接报错。这里的逻辑是幂等需要顺序保证顺序保证要求单链接排队吞吐靠批量而不是并发请求。3.2 同步发送与异步发送的取舍sarama提供两种生产者接口SyncProducer和AsyncProducer。SyncProducer的SendMessage方法会阻塞等待broker返回结果好处是代码简单、错误处理直接坏处是吞吐有上限。每次发送都有一次网络往返即使Kafka批量写了多条单线程同步发送还是会成为瓶颈。AsyncProducer的Inputchannel只管往里塞消息发送是异步的结果从Successes和Errors两个channel读取。这种模式吞吐高但需要自己处理回调代码复杂度上一个台阶。下面是一个基于AsyncProducer的封装示例type AsyncProducer struct { producer sarama.AsyncProducer } func NewAsyncProducer(brokers []string, config *sarama.Config) (*AsyncProducer, error) { p, err : sarama.NewAsyncProducer(brokers, config) if err ! nil { return nil, err } ap : AsyncProducer{producer: p} go ap.handleResults() return ap, nil } func (ap *AsyncProducer) Send(topic string, key []byte, value []byte) { msg : sarama.ProducerMessage{ Topic: topic, Key: sarama.ByteEncoder(key), Value: sarama.ByteEncoder(value), } ap.producer.Input() - msg } func (ap *AsyncProducer) handleResults() { for { select { case success : -ap.producer.Successes(): // 这里可以打日志或上报指标记录消息发送成功 _ success case errMsg : -ap.producer.Errors(): // 注意拿到的是ProducerError里面包含原始msg和err log.Printf(kafka send failed, topic%s, err%v, errMsg.Msg.Topic, errMsg.Err) // 重试逻辑不建议在这里做长阻塞最好丢回队列或进入重试topic } } }使用异步生产者时一个小技巧是不要让回调处理逻辑太重。handleResults这个goroutine如果阻塞了可能导致Successes/Errorschannel读不及时进而影响整个发送链路。回调里最合理的做法是计数、记录日志把真正的补偿逻辑交给另外的队列或任务系统。我见过有项目在Errors回调里直接睡5秒再重发结果Kafka瞬时抖动时回调goroutine全部卡住Inputchannel写不进去整个上游服务被拖垮。这种问题排查起来非常隐蔽因为代码看起来逻辑没错但就是性能直线下降。正确的做法发失败的消息落到一个本地队列或者单独的retry-topic由独立线程去处理。3.3 分区与消息key的处理Kafka的分区策略简单说就是如果消息没有key轮询或随机选择分区如果有key对key做哈希具体是murmur2或DefaultHash然后映射到分区。sarama里对应的配置是config.Producer.Partitioner。默认实现会保证同一个key落到同一个分区但要注意一个细节分区数变化之后同一个key可能落到不同分区因为哈希取模的模数变了。业务上经常需要“同一业务id的消息严格按照顺序处理”比如订单状态变更、用户行为轨迹。这种情况一定要设置key并且保证topic分区数在业务生命周期内不调整。如果分区数变了同一个key的旧消息和新消息分配到不同分区就失去了顺序性而且已经消费的历史数据也无所谓关键是新老数据交替期的顺序问题几乎无解。我自己写业务时有一个习惯Key不要传复杂结构尽量传业务id的字节串并且保持稳定。比如订单事件用订单id做key用户事件用用户id做key。这样同一订单的所有事件一定进同一个分区消费端可以放心做本地状态合并。给一个指定分区的例子虽然实际中很少手动指定分区但测试时会用到msg : sarama.ProducerMessage{ Topic: order-events, Key: sarama.StringEncoder(orderId), Value: sarama.StringEncoder(eventBody), Partition: 0, // 手动指定分区一般不建议 }手动指定分区的坏处很明显某个分区数据量就会特别大其他分区闲着违背了分区均衡的目的。所以除非有明确的局部性需求否则还是交给partitioner。3.4 发送结果回调与错误处理生产端最容易出的两类错误一类是网络层面的比如连接超时、broker不可达另一类是broker返回的错误比如TopicAuthorizationException、NotLeaderForPartitionException。sarama的Retry配置会自动处理一部分错误比如分区leader变更导致的NotLeader错误客户端刷新metadata后会重试。但重试不是万能的主题不存在、权限不足这类错误重试多少次都白搭所以重试次数不要设成无限大。我一般设Retry.Max 3到5再多其实没有意义只会拖长故障恢复时间。对于同步生产者SendMessage返回error后消息是失败了还是成功但你不知道这取决于具体错误类型。比如超时错误可能是已经写入但响应超时也就是“不确定状态”。这种情况下如果没有幂等业务上的表现就是“可能重复”所以生产环境开启幂等不是一个可选项而是必须项。对于异步生产者错误都从Errors()channel出来。ProducerError结构体里同时有Msg和Err记录日志时建议把Msg.Topic、Msg.Partition、Msg.Offset都打出来。如果msg的offset为0且partition为0说明这条消息从未被分配分区错误发生在发送前这种情况重试可以直接重新构造消息。4. 消费者消费组、offset与重平衡实践4.1 消费者配置的陷阱消费者的初始化配置比生产者还容易踩坑。看这段基础配置func newConsumerGroup(groupID string) (sarama.ConsumerGroup, error) { config : sarama.NewConfig() config.Version sarama.V3_5_0_0 // 没有提交过offset时从最老的消息开始消费 config.Consumer.Offsets.Initial sarama.OffsetOldest // 消费组会话超时。如果consumer在超时时间内没有心跳会被认为死亡并触发重平衡 config.Consumer.Group.Session.Timeout 20 * time.Second // 心跳间隔必须小于Session.Timeout config.Consumer.Group.Heartbeat.Interval 4 * time.Second // 重平衡策略 config.Consumer.Group.Rebalance.GroupStrategies []sarama.BalanceStrategy{sarama.NewBalanceStrategyRange()} // 我自己习惯使用手动提交这里关闭自动提交 config.Consumer.Offsets.AutoCommit.Enable true // 自动提交间隔如果开启自动提交的话 config.Consumer.Offsets.AutoCommit.Interval 5 * time.Second // 返回错误到Errors() channel config.Consumer.Return.Errors true client, err : sarama.NewConsumerGroup([]string{localhost:9092}, groupID, config) if err ! nil { return nil, err } return client, nil }几个非常关键的点Initial的取值OffsetNewest表示只消费启动之后新到达的消息OffsetOldest表示从最早的消息开始。如果业务上要求消息不丢一般用OffsetOldest如果只关心实时增量用OffsetNewest。很多刚接触Kafka的人在这上面栽跟头测试时用OffsetNewest生产了一条消息后发现消费不到就是因为客户端在消息到达之前启动它认为没有旧消息需要处理。Session.Timeout和Heartbeat.Interval这两个参数必须配套设置。心跳间隔一般设为会话超时的五分之一到四分之一这样在会话超时前至少能发几轮心跳。如果心跳间隔大于或等于会话超时consumer会在没有异常的情况下不断触发重平衡表现为“消费几秒就停止过一会又恢复”日志里全是重平衡记录。组内消费者数量与分区数消费组拿到的分区数是min(组内成员数, 分区数)。如果分区数是3组里起了6个consumer那有3个consumer会一直空闲分配不到任何分区。这个现象很正常但新手容易误以为是负载不均衡。4.2 消费组与offset提交策略sarama消费组的使用套路是实现ConsumerGroupHandler接口然后在Consume循环里持续监听。type OrderHandler struct{} func (h OrderHandler) Setup(session sarama.ConsumerGroupSession) error { log.Println(consumer group rebalanced, setup) return nil } func (h OrderHandler) Cleanup(session sarama.ConsumerGroupSession) error { log.Println(consumer group rebalanced, cleanup) return nil } func (h OrderHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { // 这里会收到被分配给当前实例的分区消息 for msg : range claim.Messages() { // 1. 处理消息 processMessage(msg) // 2. 手动提交offset标记这条消息已处理完成 session.MarkMessage(msg, ) } return nil } func runConsumer() { handler : OrderHandler{} client, _ : newConsumerGroup(order-group) defer client.Close() // 使用通道控制循环退出 ctx, cancel : context.WithCancel(context.Background()) defer cancel() for { err : client.Consume(ctx, []string{order-events}, handler) if err ! nil { log.Printf(consume error, err%v, err) time.Sleep(5 * time.Second) continue } // 如果ctx被取消说明需要退出了 if ctx.Err() ! nil { break } // 如果没有错误返回说明发生了重平衡触发handler.Setup/Cleanup // 这里尽量不要直接break应该回到for循环继续等待下一轮消费 } }这里有个很关键的行为client.Consume是一个阻塞操作当消费组发生重平衡时它可能会从阻塞中返回此时handler的Setup和Cleanup方法被调用。代码里要用for {}循环让消费在重平衡结束后继续而不是退出。offset提交有自动和手动两种。自动提交的情况AutoCommit.Enable true时sarama会每隔AutoCommit.Interval自动提交session.MarkMessage标记过的offset。注意是“标记过的”不是“读到的”。如果业务处理一条消息耗时很长消息还在处理中自动提交的可能是之前的offset。比如分区里已经有1-10号消息处理到5号时自动提交把“4号已处理”假设每次循环Mark的offset提交上去然后服务宕机重启后从5号开始消费4号会被重复消费一次。手动提交的情况关闭自动提交在处理完一批消息后主动调用session.Commit()。这种方式把提交时机完全掌握在业务方手里代价是如果忘记提交重启后会重新消费很多消息。我个人推荐生产环境用手动提交。方法// 在ConsumeClaim里每处理完一条消息就Mark session.MarkMessage(msg, ) // 每处理完固定数量或者固定时间间隔手动提交一次 if count%100 0 { if err : session.Commit(); err ! nil { log.Printf(commit offset failed, err%v, err) // 提交失败会导致offset未持久化重启后重复消费 // 这里可以选择阻塞重试或者记录日志后继续但要接受重复消费风险 } }手动提交的核心思想先Mark后Commit。Mark只是内存状态Commit才是真正发送到broker。如果进程在Mark之后、Commit之前崩溃offset不会推进恢复后依然从旧的offset开始读消息会被重复消费。这是分布式系统里“最少一次”at least once语义的必然结果能保证不丢但不能保证不重。要彻底解决重复需要在业务侧做幂等。4.3 手动提交与幂等消费“消息处理成功”之后才能Mark这句话看似废话但很多人写成“先Mark再处理”。如果处理失败offset已经被标记消息就丢了一部分。正确写法for msg : range claim.Messages() { if err : processMessage(msg); err ! nil { // 处理失败可以选择阻塞重试、跳过并记日志、或者放到重试队列 // 注意不要Mark失败的消息否则就相当于把这条消息丢掉了 retryMessage(msg) continue // 或者根据业务决定是否停止当前分区消费 } session.MarkMessage(msg, ) }这里的继续处理其实带着风险如果某条消息一直处理失败你会陷入“继续消费下一条失败消息永远不提交”的困境。更严重的消息处理慢整个分区的消费速度下降会引发消费堆积。幂等消费的通用方案在消息体里带一个唯一业务id消费端在处理前去查一下“这个id是否已处理过”。实现方式可以用数据库唯一索引、Redis的SETNX、或者本地布隆过滤器没有通用答案得看业务存储。关键是保证“查重写业务数据”这两个动作在一个事务或一个原子操作里完成否则还是会撞上并发重复。4.4 重平衡期间的消费保护消费组重平衡是线上事故的高发场景。当一个consumer实例退出或加入时分区会重新分配。在你旧的consumer处理完最后一批消息之前Kafka可能已经把这个分区分配给了新的consumer此时两个consumer同时处理同一分区业务上就会出现重复。sarama的ConsumerGroupHandler提供了Setup和Cleanup钩子语义上Setup重平衡完成后当前consumer期望开始消费前被调用。可以在这里初始化资源、清空缓存状态。Cleanup消费者释放某个分区前被调用。可以在这里做最后提交、关闭资源。需要注意Cleanup并不保证在这个consumer里被调用的时机绝对安全只要重平衡发生任何实例都可能被要求释放分区。如果业务上有依赖分区的本地状态比如要把某个ID的最近一条消息状态放在内存里要意识到重平衡时这个状态会失效不能假设分区永远在自己手里。一个很实际的建议不要在ConsumeClaim里处理需要跨消息维护大量状态的任务。比如依赖“同一订单的所有事件都在我这个实例上”重平衡一发生这个假设就破了。如果业务对顺序有强依赖最好是在下游数据库、状态机做最终一致性而不是把宝押在“分区一直归我”上。5. 常见问题与排查技巧实录5.1 连接失败与版本兼容现象1producer初始化成功但发送消息时报kafka: client has run out of available brokers to talk to。排查思路先用命令行工具验证broker地址可不可达。很多坑是Kafka容器里的ADVERTISED_LISTENERS配成了容器内网IP宿主机上的Golang程序拿着那个地址肯定连不上。把ADVERTISED_LISTENERS改成localhost:9092或者宿主可达的地址。现象2kafka: client/metadata got error from broker ... (unrecognized broker version)这类错误。基本是config.Version没设置或设太低。Kafka broker 3.x的协议和2.x差异很大尤其是消费组协调协议。把config.Version明确设成broker版本附近的值问题会消失。现象3升级sarama大版本后原来正常的发送变得超时。sarama从Shopify迁移到IBM后部分配置默认值改过。Producer.Return.Successes默认值是false如果你用的是SyncProducer却忘了配置Return.Successes trueSendMessage永远返回一个“成功”但其实没拿到结果的状态。这个问题很坑因为代码不报错只是行为不符合预期。检查config一眼看过去全是对的话重点查这个字段。5.2 重复消费与offset丢失现象服务正常消费但每次重启后都有一部分消息重复消费。优先检查是不是自动提交间隔大于消费到提交之间的时间差。如果业务处理很慢自动提交严重滞后重启自然重复。解决办法开启手动提交在业务处理成功后立刻Commit。现象多个consumer实例同时运行日志里出现同一条消息被重复处理的记录。有可能是组内重平衡时旧实例的Cleanup没来得及提交最后的offset新实例从旧offset开始消费。这种场景下保证Cleanup里调用一次Commit能减少窗口但没法根除。根除还是要靠业务幂等。我踩过的一个具体场景消费组用默认的range分配策略RangeAssignortopic有6个分区group里有3个实例暂时均匀。后来新增一个实例到4个Range策略重新分配后部分实例的分区数变化很大挨着被换边的分区最容易出现重复消费窗口。解决策略是换成StickyAssignor它尽量保持原有分配减少分区迁移。这个优化在消费实例频繁变更时效果很明显。5.3 消费堆积与性能瓶颈消费堆积是下游处理能力不足的典型表现。排查时先区分是“单条处理慢”还是“单位时间处理条数低”。单条处理慢可能消息里带了大对象反序列化耗时也可能处理逻辑里碰了数据库慢查询。定位方法是在processMessage里加耗时日志找出p99耗时。单位时间处理条数低检查是不是分区数太少导致消费者数量上不去。举例topic分区数是3下游即使起了20个实例实际消费的也只有3个实例。扩容分区是表象解法更合理的是一开始根据预估吞吐量设计分区数。另外确认一下是否批量处理。sarama的ConsumerGroupClaim.Messages()返回的是一个channel一次循环只能拿到一条。想批量的话可以在循环里收集N条再统一处理减少数据库连接建立等固定开销。给一个攒批消费的简化思路func (h BatchHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { batch : make([]*sarama.ConsumerMessage, 0, batchSize) for msg : range claim.Messages() { batch append(batch, msg) if len(batch) batchSize { if err : processBatch(batch); err ! nil { // 处理失败可以整体重试或部分重试 return err // 注意返回err会让Consume退出需要外层恢复 } for _, m : range batch { session.MarkMessage(m, ) } session.Commit() batch batch[:0] } } return nil }这种攒批模式能显著提升吞吐但代价是消息可靠性和响应延迟之间的平衡攒得越多单条消息的处理延迟越高故障时丢失窗口越大。5.4 排查命令速查场景有用命令查看topic分区和leaderkafka-topics.sh --bootstrap-server localhost:9092 --describe --topic order-events查看消费组当前offsetkafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group order-group查看消费组堆积滞后量同上看LAG列从指定offset开始消费kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic order-events --partition 0 --offset 100生产一条测试消息kafka-console-producer.sh --bootstrap-server localhost:9092 --topic order-events排查顺序一般是先看topic有没有数据再看消费组offset情况最后才怀疑代码。很多时候不是代码问题而是消息根本没进到预期的topic。6. 上线前后的调优与观测6.1 参数调优组合给一组基于中等流量场景峰值5K TPS、单条消息1KB的经验配置不是绝对标准但是一个合理的起点。生产者侧参数建议值说明Version与broker对齐避免协议降级RequiredAcksWaitForAll保证多副本确认Idempotenttrue防止重试导致重复Retry.Max3~5过多重试无意义Flush.Messages100~500攒够条数再发Flush.Bytes64KB~1MBbatch字节数阈值Flush.Frequency10~20ms延迟与吞吐的平衡Compressionsnappy或zstd压缩比高、CPU开销可接受消费者侧参数建议值说明InitialOffsetOldest避免漏消息Session.Timeout20~30s允许短暂GC停顿Heartbeat.Interval4~6s小于Timeout即可AutoCommit.Enablefalse手动提交更可控MaxProcessingTime视业务而定不影响fetch但影响处理模型Flush.Frequency这个参数很多人不设置结果生产者端不攒批一条一条发。吞吐上不去时第一件事就是检查batch相关配置有没有生效。日志里如果看到很多空转或者大量小请求基本就是攒批没配好。6.2 压测与监控写完了生产者和消费者不要直接上线上流量。先在测试环境做简单压测验证几个核心指标。压测脚本可以直接用sarama的AsyncProducer疯狂发消息观察这几项发送成功率和异常类型重点看Errorschannel里是不是有MessageSizeTooLarge或超时类错误。消费端延迟给每条消息带上发送时间消费者处理时计算当前时间与发送时间的差值。LAG值压测结束后用kafka-consumer-groups.sh看消费组的LAG是否归零。我常用的做法先固定分区数为12消费者4个实例逐步加大发送速率观察消费端延迟曲线的拐点。如果延迟从几百毫秒突然涨到几秒说明消费处理能力到了瓶颈需要横向加实例或者优化单条处理逻辑。加实例时注意实例数超过分区数后不会再有提升。监控这块几点经验每发送一条消息成功记录一个counter失败时记录另一个counter。看趋势不要只看绝对值。消费端记录“处理耗时”和“处理消息数”每30秒打一次日志或上报一次。定时检查消费组LAG这是最直接的消费健康度指标。6.3 消息生产与消费的业务侧检查清单上线前过一遍这些问题能拦下不少问题key是否稳定分区数变动是否会影响业务顺序需求消息体是否可能超过broker的message.max.bytes限制消费端处理失败时是否有重试策略和丢弃策略是否做了消息幂等消费组实例数是否小于分区数导致部分机器空转消费者启动时是OffsetOldest还是OffsetNewest是否符合预期这一套检查完基本可以放心地把sarama生产者消费者接进业务链路了。最后分享一个我自己在多次生产事故里体会最深的原则Kafka能保证消息不丢但很难保证消息不重能保证分区内有序但很难保证跨分区全局有序。所有系统设计都要围绕这两个现实来做。消费端幂等、业务侧接受至少一次语义是比任何客户端参数都更重要的一道防线。另一个小技巧是新consumer上线前先让它在测试环境跑一段时间专门观察重平衡Trigger日志因为重平衡是否频繁直接决定了你的是不是会在半夜接到报警电话。sarama这套库本身功能足够完整只要你把offset提交、版本兼容、重平衡这三个关键点稳住后面的路就会顺畅很多。