1. 为什么要在 Scala 里手写 Kafka 生产消费实例1.1 从一次数据管道抖动说起前阵子帮一个做实时风控的团队排查问题他们的数据链路是 Flink 作业消费上游埋点处理完再写回 Kafka 供下游规则引擎使用。现象很怪白天流量高峰期下游规则引擎偶尔会收到重复消息凌晨低峰期又一切正常。查了两天最后定位到问题出在一段 Scala 写的 Kafka 生产者代码上——acks配的是1重试次数设了Integer.MAX_VALUE但没开幂等。高峰期 broker 主从切换消息写进去了但 ack 超时生产者重试于是下游就看到了重复。这件事让我意识到一个很现实的问题现在大家写 Kafka 客户端要么用 Spring Kafka 一把梭要么直接上 Flink Connector真正手写KafkaProducer/KafkaConsumer的场景反而少了。但一旦出问题恰恰是这些被封装掉的细节最要命。Scala 这门语言在 Kafka 生态里位置很特殊——Kafka 本身就是 Scala 写的官方客户端kafka-clients是 Java 库Scala 调用它天然无缝同时 Scala 的函数式特性又能把生产消费的样板代码压得很薄。这篇东西就是把我这些年用 Scala 写 Kafka 生产消费实例的经验整理出来。核心关键词是 Scala、Kafka、生产者、消费者、幂等、手动提交、背压。内容会从依赖选型讲到完整可跑的实例再到参数调优和踩坑排查。适合两类人看一是刚接触 Kafka、想搞明白客户端到底在干什么的 Scala 开发者二是用惯了封装框架、想回头补一补底层细节的老手。我不会只贴代码每个参数为什么这么设、不这么设会出什么事都会讲清楚。1.2 先明确一件事Scala 调 Kafka 到底调的是什么很多人有个误解以为 Scala 有专门的 Kafka 客户端。其实没有。Kafka 官方提供的客户端就是org.apache.kafka:kafka-clients这个 Java 包Scala 项目直接引它就行。所谓Scala 实现 Kafka 生产消费本质是用 Scala 的语法糖去包装 Java 客户端 API再加上 Scala 生态里的一些惯用法比如用Future做异步发送、用case class做消息体序列化、用Try处理发送结果。这个认知很重要因为它决定了你排查问题的方向。当你遇到序列化异常、超时、重平衡这些问题时根因几乎都在 Java 客户端那一层Scala 只是调用方。所以别指望 Scala 能帮你绕过 Kafka 的固有机制它只能让你写得更舒服。那 Scala 相比直接写 Java 的优势在哪我总结三点。第一类型推导和样例类让消息定义极其简洁一个case class UserEvent(...)配合 Circe 或 Play JSON 就能搞定序列化比 Java 里写一堆 getter/setter 清爽太多。第二函数式组合让异步发送的回调逻辑可以用map/flatMap串起来不用嵌套一堆匿名内部类。第三模式匹配处理消费到的消息特别顺手尤其是多种消息类型混在一个 topic 里的场景。2. 环境准备与依赖选型的关键决策2.1 依赖版本怎么选才不踩坑先上依赖。一个典型的 Scala Kafka 项目build.sbt里核心就这几行libraryDependencies Seq( org.apache.kafka % kafka-clients % 3.6.1, com.typesafe.play %% play-json % 2.10.4, org.slf4j % slf4j-api % 2.0.9, ch.qos.logback % logback-classic % 1.4.14 )这里有几个决策点值得说。kafka-clients的版本要和 broker 版本对齐但不是严格相等。Kafka 客户端有很好的向后兼容性3.6 的客户端连 2.8 的 broker 一般没问题但反过来用 2.4 的客户端连 3.6 的 broker 就可能缺一些新协议支持。我的经验是客户端版本不低于 broker 大版本减一。比如 broker 是 3.6客户端用 3.4 以上都稳。JSON 库的选择是个容易被忽视的点。Play JSON 胜在和 Scala 集成好、Reads/Writes用起来自然Circe 更纯粹但学习曲线陡一点Jackson 性能最好但 Scala 味不足。我选 Play JSON 是因为它在够用和好写之间平衡得最好尤其是配合case class的隐式转换几行就能定义好序列化器。注意kafka-clients是纯 Java 库用%而不是%%引入别手滑写成%%否则 SBT 会去找 Scala 版本后缀的包直接报找不到。2.2 序列化器的选型逻辑Kafka 消息的 key 和 value 都需要序列化器。内置的有StringSerializer、ByteArraySerializer、IntegerSerializer等。但实际项目里消息体通常是复杂对象你得自己实现Serializer[T]和Deserializer[T]。这里有个关键决策序列化器要不要带 schema 信息。纯 JSON 序列化简单但下游解析时如果字段变了容易炸Avro 或 Protobuf 带 schema 演进能力但引入 Schema Registry 又增加运维成本。我的建议是内部系统、字段相对稳定的场景JSON 足够跨团队、字段频繁演进的场景老老实实上 Avro。下面是一个基于 Play JSON 的通用序列化器实现这个我用了好几年很稳import org.apache.kafka.common.serialization.{Deserializer, Serializer} import play.api.libs.json.{Format, Json} class JsonSerializer[T](implicit fmt: Format[T]) extends Serializer[T] { override def serialize(topic: String, data: T): Array[Byte] if (data null) null else Json.toJson(data).toString.getBytes(UTF-8) } class JsonDeserializer[T](implicit fmt: Format[T]) extends Deserializer[T] { override def deserialize(topic: String, data: Array[Byte]): T if (data null) null.asInstanceOf[T] else Json.parse(new String(data, UTF-8)).as[T] }注意serialize里对null的处理。Kafka 允许消息 value 为 null表示 tombstone用于日志压缩场景删除 key如果你的序列化器不处理 null遇到 tombstone 就会 NPE。这个坑我见过不止一个团队踩。3. 生产者实例从能跑到跑得稳3.1 最小可用生产者与参数逐条拆解先给一个能跑的最小生产者然后逐条讲参数import org.apache.kafka.clients.producer._ import org.apache.kafka.common.serialization.StringSerializer import java.util.Properties object SimpleProducer { def main(args: Array[String]): Unit { val props new Properties() props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092) props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName) props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName) props.put(ProducerConfig.ACKS_CONFIG, all) props.put(ProducerConfig.RETRIES_CONFIG, 3) props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true) val producer new KafkaProducer[String, String](props) try { val record new ProducerRecord[String, String](user-events, user-1, login) val metadata producer.send(record).get() println(s发送成功: partition${metadata.partition()}, offset${metadata.offset()}) } finally { producer.close() } } }acksall意味着 leader 要等所有 ISRin-sync replicas副本都写入才返回成功。这是不丢消息的前提代价是延迟略高。如果设成1leader 写完就返回leader 挂了但副本还没同步消息就丢了。设成0更是发出去就不管性能最高但基本等于赌博。retries3配合enable.idempotencetrue才是正确姿势。幂等生产者是 Kafka 0.11 引入的关键特性它给每条消息分配序列号broker 端去重保证重试不会产生重复消息。但幂等有前提acks必须是allretries必须大于 0max.in.flight.requests.per.connection不能超过 5。这几个条件不满足幂等自动失效而且 Kafka 不会报错只是静默降级——这就是我开头那个案例的根因。3.2 异步发送与回调的正确写法同步send().get()会阻塞吞吐上不去。生产环境基本都用异步加回调producer.send(record, new Callback { override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit { if (exception ! null) { // 这里千万别只打日志就完事 logger.error(s发送失败: topic${record.topic()}, exception) } else { logger.debug(soffset${metadata.offset()}) } } })用 Scala 可以写得更优雅把回调包成Futureimport scala.concurrent.{Future, Promise} def sendAsync[K, V](producer: KafkaProducer[K, V], record: ProducerRecord[K, V]): Future[RecordMetadata] { val promise Promise[RecordMetadata]() producer.send(record, (metadata: RecordMetadata, exception: Exception) { if (exception ! null) promise.failure(exception) else promise.success(metadata) }) promise.future }实操心得回调里绝对不要做耗时操作。Kafka 生产者的回调是在 sender 线程里执行的你在回调里查数据库、发 HTTP 请求会直接阻塞整个发送管道吞吐断崖式下跌。要处理失败消息正确做法是丢到一个内存队列由单独的线程消费。3.3 分区策略与消息顺序的取舍消息发到哪个 partition由partitioner.class决定。默认策略是有 key 就按 key 的 hash 取模没 key 就轮询2.4 之后改成了 sticky partitioner攒一批再换分区减少小批次。顺序性只在单个 partition 内保证。如果你的业务要求同一用户的事件严格有序就必须用 userId 做 key让同一用户的消息落到同一分区。但这样会带来数据倾斜风险——某个大 V 用户的消息量可能是普通用户的几万倍全压在一个分区上那个分区的 broker 就成了瓶颈。我处理过一个案例某活动期间头部几个用户的消息占了 40% 流量全挤在三个分区上消费端 lag 直接飙到百万。后来改成userId 时间窗口做 key牺牲了跨窗口的顺序性换来了负载均衡。所以顺序性和吞吐量往往要权衡没有银弹。4. 消费者实例提交策略决定一切4.1 消费者组与重平衡机制消费者最核心的概念是消费者组consumer group。同一个组内的消费者分摊分区一个分区只能被组内一个消费者消费。这个约束决定了消费并行度的上限就是分区数——你开 10 个消费者但 topic 只有 3 个分区那 7 个消费者就是白占资源。重平衡rebalance是消费者最头疼的事。触发条件有三个消费者加入、消费者离开、分区数变化。重平衡期间整个组会暂停消费eager 协议下如果频繁触发消费延迟会非常明显。Kafka 2.4 引入了增量协作重平衡CooperativeStickyAssignor重平衡时只迁移必要的分区其他消费者继续工作。这个改进很大新项目建议直接用props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, org.apache.kafka.clients.consumer.CooperativeStickyAssignor)4.2 手动提交 vs 自动提交一个必须想清楚的选择自动提交enable.auto.committrue看起来省事但有个致命问题它是在 poll 循环里按时间间隔提交的提交的是上次 poll 返回的 offset。如果你 poll 了一批消息处理到一半崩了自动提交可能已经把 offset 提交了那批没处理完的消息就永远丢了。手动提交才是生产环境的正确选择。手动提交又分两种提交方式时机风险适用场景commitSync处理完后同步提交阻塞吞吐低消息重要、不能丢commitAsync处理完后异步提交失败不重试可能重复吞吐优先、能容忍重复同步异步混合常规异步关闭前同步兼顾两者大多数生产场景混合提交是我最推荐的模式代码长这样val consumer new KafkaConsumer[String, String](props) consumer.subscribe(Collections.singletonList(user-events)) try { while (running) { val records consumer.poll(Duration.ofMillis(500)) records.forEach { record processRecord(record) // 业务处理 } if (!records.isEmpty) { consumer.commitAsync() // 常规异步提交 } } } finally { try { consumer.commitSync() // 关闭前同步提交确保不丢 } finally { consumer.close() } }commitAsync失败不重试是有意设计的——如果异步提交失败后重试可能提交的是更旧的 offset反而覆盖了更新的提交。所以常规用异步只在关闭时用同步兜底。4.3 消费位移与重复消费的本质很多人搞不清至少一次和最多一次到底怎么来的。其实就一句话先提交 offset 再处理消息就是最多一次可能丢先处理消息再提交 offset就是至少一次可能重复。Kafka 默认走的是至少一次。这意味着你的业务处理逻辑必须幂等。怎么做到幂等常见几种用消息里的唯一 ID 做去重表处理前先查用数据库的唯一索引重复插入直接失败用 Redis 的SETNX做短期去重窗口我一般推荐第一种虽然多一次查询但最可靠。去重表的清理策略要设计好比如保留 7 天用定时任务删过期数据。5. 完整可跑的实例一个端到端的消息管道5.1 消息模型定义把前面所有东西串起来做一个完整的例子模拟用户行为事件的生产和消费。import play.api.libs.json.{Format, Json} case class UserEvent( eventId: String, userId: String, eventType: String, timestamp: Long, payload: Map[String, String] ) object UserEvent { implicit val format: Format[UserEvent] Json.format[UserEvent] }eventId是幂等去重的关键生产端生成 UUID消费端用它查重。5.2 生产端完整实现import org.apache.kafka.clients.producer._ import java.util.Properties import java.util.UUID import scala.util.{Try, Success, Failure} class UserEventProducer(bootstrapServers: String) { private val props new Properties() props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers) props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName) props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[JsonSerializer[UserEvent]].getName) props.put(ProducerConfig.ACKS_CONFIG, all) props.put(ProducerConfig.RETRIES_CONFIG, 5) props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true) props.put(ProducerConfig.LINGER_MS_CONFIG, 10) props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768) props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, lz4) private val producer new KafkaProducer[String, UserEvent](props) def send(event: UserEvent): Try[RecordMetadata] { val record new ProducerRecord[String, UserEvent]( user-events, event.userId, event ) Try(producer.send(record).get()) } def close(): Unit producer.close() }linger.ms10和batch.size32768是吞吐优化的关键。生产者不会来一条发一条而是攒够batch.size或等够linger.ms才发一批。linger.ms从默认 0 调到 10吞吐能提升好几倍代价是最多 10ms 的额外延迟。这个取舍要看业务实时性要求极高的场景别调太大。compression.typelz4是性价比最高的压缩算法压缩比不错CPU 开销低。gzip 压缩比更高但慢snappy 快但压缩比一般。lz4 是大多数场景的甜点。5.3 消费端完整实现import org.apache.kafka.clients.consumer._ import java.util.{Collections, Properties} import java.time.Duration class UserEventConsumer(bootstrapServers: String, groupId: String) { private val props new Properties() props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers) props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId) props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getName) props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[JsonDeserializer[UserEvent]].getName) props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false) props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 200) props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000) props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, org.apache.kafka.clients.consumer.CooperativeStickyAssignor) private val consumer new KafkaConsumer[String, UserEvent](props) volatile private var running true def start(): Unit { consumer.subscribe(Collections.singletonList(user-events)) try { while (running) { val records consumer.poll(Duration.ofMillis(500)) records.forEach { record processWithRetry(record.value()) } if (!records.isEmpty) consumer.commitAsync() } } finally { try consumer.commitSync() finally consumer.close() } } private def processWithRetry(event: UserEvent): Unit { // 幂等处理先查 eventId 是否已处理 if (!isProcessed(event.eventId)) { doProcess(event) markProcessed(event.eventId) } } def shutdown(): Unit running false }max.poll.records200和max.poll.interval.ms300000这两个参数要配合着看。如果一批消息处理时间超过max.poll.interval.ms消费者会被踢出组触发重平衡。所以要么调大这个值要么调小max.poll.records让每批处理更快。我一般根据单条处理耗时反推单条 50ms一批 200 条就是 10 秒max.poll.interval.ms设 5 分钟绰绰有余。6. 常见问题排查与避坑清单6.1 生产端典型问题速查现象可能原因排查方向发送超时TimeoutExceptionbroker 不可达或网络抖动检查bootstrap.servers、防火墙、DNS消息重复幂等未生效检查 acks/retries/in.flight 三件套吞吐上不去linger.ms 太小、批次太小调大 linger 和 batch.sizeRecordTooLargeException单条消息超过max.request.size调大参数或拆分消息序列化异常序列化器与消息类型不匹配检查泛型和隐式 Format6.2 消费端典型问题速查现象可能原因排查方向频繁重平衡处理超时或心跳超时调大 max.poll.interval.ms、session.timeout.ms消费 lag 持续增长消费能力不足加消费者不超过分区数、优化处理逻辑重复消费提交前崩溃业务幂等 合理提交策略消息丢失自动提交 处理失败改手动提交分区分配不均默认分配器不均衡换 CooperativeStickyAssignor6.3 几个只有踩过才知道的坑第一个坑close()不调用会丢消息。生产者有缓冲区send()返回不代表消息发出去了。程序退出前不调close()缓冲区里的消息全丢。我见过有人用System.exit(0)直接退缓冲区消息一条没发出去。正确做法是注册 shutdown hook或者用 try-finally 保证 close。第二个坑消费者线程不安全。KafkaConsumer不是线程安全的多线程共享一个实例会抛ConcurrentModificationException。要么一个线程一个消费者要么用单线程 poll、多线程处理的模式但提交 offset 要回到 poll 线程做。第三个坑wakeup()是唯一安全的跨线程中断方式。想优雅停止消费者别用Thread.interrupt()用consumer.wakeup()它会让poll()抛WakeupException你在 catch 里处理完再 close。第四个坑topic 自动创建是个陷阱。auto.create.topics.enabletrue时生产者往不存在的 topic 发消息会自动创建但分区数和副本数用的是 broker 默认值往往不是你想要的。生产环境建议关掉自动创建topic 由运维统一管理。7. 性能调优的取舍逻辑7.1 吞吐、延迟、可靠性三角Kafka 调优本质是在吞吐、延迟、可靠性三者之间找平衡点不可能同时最优。追求吞吐调大batch.size、linger.ms、compression.type消费端调大fetch.min.bytes、max.poll.records。追求低延迟linger.ms0、batch.size调小消费端fetch.min.bytes1。追求可靠acksall、enable.idempotencetrue、手动提交、min.insync.replicas2。我的经验是先明确业务底线再优化其他维度。金融交易类可靠性是底线吞吐和延迟都得让路日志采集类吞吐是底线偶尔丢几条无所谓。7.2 一个真实的调优过程之前有个日志采集场景单机生产吞吐只有 5 万条/秒目标是 20 万。我按这个顺序调第一步linger.ms从 0 调到 20吞吐直接翻倍到 11 万。因为原来每条消息一个请求网络往返开销巨大。第二步batch.size从 16KB 调到 128KB吞吐到 15 万。批次大了压缩效率也上来了。第三步compression.type从 none 改成 lz4吞吐到 19 万。压缩减少了网络传输量。第四步buffer.memory从 32MB 调到 64MB吞吐稳定在 21 万。缓冲区不够时send()会阻塞调大后发送更顺畅。整个过程没动 broker 配置纯客户端调优就达标了。所以别一上来就怀疑 broker客户端参数往往才是瓶颈。8. 写在最后的一点个人体会用 Scala 写 Kafka 生产消费语言本身带来的便利是次要的真正重要的是对 Kafka 客户端机制的理解。我见过太多人把KafkaProducer当成一个简单的send方法用参数全默认出了问题就抓瞎。其实每一个参数背后都是 Kafka 设计者对某种场景的权衡你理解了权衡才知道自己的场景该往哪边靠。如果让我给刚上手的人一条建议那就是先把acks、enable.idempotence、enable.auto.commit这三个参数彻底搞明白。它们分别对应生产不丢、生产不重、消费不丢是整条链路可靠性的地基。地基打牢了再谈吞吐和延迟优化才有意义。至于 Scala 的那些语法糖用熟了自然就顺手。Future包异步发送、case class定义消息、模式匹配处理多类型消息这些写起来确实比 Java 舒服。但记住舒服归舒服底层还是那套 Java 客户端出问题还得回到那一层去找答案。 
长河君 · 2026/10/17 1:36:12 · 阅读 1.2k Scala 手写 Kafka 生产消费实例：幂等、手动提交与背压调优
1. 为什么要在 Scala 里手写 Kafka 生产消费实例1.1 从一次数据管道抖动说起前阵子帮一个做实时风控的团队排查问题他们的数据链路是 Flink 作业消费上游埋点处理完再写回 Kafka 供下游规则引擎使用。现象很怪白天流量高峰期下游规则引擎偶尔会收到重复消息凌晨低峰期又一切正常。查了两天最后定位到问题出在一段 Scala 写的 Kafka 生产者代码上——acks配的是1重试次数设了Integer.MAX_VALUE但没开幂等。高峰期 broker 主从切换消息写进去了但 ack 超时生产者重试于是下游就看到了重复。这件事让我意识到一个很现实的问题现在大家写 Kafka 客户端要么用 Spring Kafka 一把梭要么直接上 Flink Connector真正手写KafkaProducer/KafkaConsumer的场景反而少了。但一旦出问题恰恰是这些被封装掉的细节最要命。Scala 这门语言在 Kafka 生态里位置很特殊——Kafka 本身就是 Scala 写的官方客户端kafka-clients是 Java 库Scala 调用它天然无缝同时 Scala 的函数式特性又能把生产消费的样板代码压得很薄。这篇东西就是把我这些年用 Scala 写 Kafka 生产消费实例的经验整理出来。核心关键词是 Scala、Kafka、生产者、消费者、幂等、手动提交、背压。内容会从依赖选型讲到完整可跑的实例再到参数调优和踩坑排查。适合两类人看一是刚接触 Kafka、想搞明白客户端到底在干什么的 Scala 开发者二是用惯了封装框架、想回头补一补底层细节的老手。我不会只贴代码每个参数为什么这么设、不这么设会出什么事都会讲清楚。1.2 先明确一件事Scala 调 Kafka 到底调的是什么很多人有个误解以为 Scala 有专门的 Kafka 客户端。其实没有。Kafka 官方提供的客户端就是org.apache.kafka:kafka-clients这个 Java 包Scala 项目直接引它就行。所谓Scala 实现 Kafka 生产消费本质是用 Scala 的语法糖去包装 Java 客户端 API再加上 Scala 生态里的一些惯用法比如用Future做异步发送、用case class做消息体序列化、用Try处理发送结果。这个认知很重要因为它决定了你排查问题的方向。当你遇到序列化异常、超时、重平衡这些问题时根因几乎都在 Java 客户端那一层Scala 只是调用方。所以别指望 Scala 能帮你绕过 Kafka 的固有机制它只能让你写得更舒服。那 Scala 相比直接写 Java 的优势在哪我总结三点。第一类型推导和样例类让消息定义极其简洁一个case class UserEvent(...)配合 Circe 或 Play JSON 就能搞定序列化比 Java 里写一堆 getter/setter 清爽太多。第二函数式组合让异步发送的回调逻辑可以用map/flatMap串起来不用嵌套一堆匿名内部类。第三模式匹配处理消费到的消息特别顺手尤其是多种消息类型混在一个 topic 里的场景。2. 环境准备与依赖选型的关键决策2.1 依赖版本怎么选才不踩坑先上依赖。一个典型的 Scala Kafka 项目build.sbt里核心就这几行libraryDependencies Seq( org.apache.kafka % kafka-clients % 3.6.1, com.typesafe.play %% play-json % 2.10.4, org.slf4j % slf4j-api % 2.0.9, ch.qos.logback % logback-classic % 1.4.14 )这里有几个决策点值得说。kafka-clients的版本要和 broker 版本对齐但不是严格相等。Kafka 客户端有很好的向后兼容性3.6 的客户端连 2.8 的 broker 一般没问题但反过来用 2.4 的客户端连 3.6 的 broker 就可能缺一些新协议支持。我的经验是客户端版本不低于 broker 大版本减一。比如 broker 是 3.6客户端用 3.4 以上都稳。JSON 库的选择是个容易被忽视的点。Play JSON 胜在和 Scala 集成好、Reads/Writes用起来自然Circe 更纯粹但学习曲线陡一点Jackson 性能最好但 Scala 味不足。我选 Play JSON 是因为它在够用和好写之间平衡得最好尤其是配合case class的隐式转换几行就能定义好序列化器。注意kafka-clients是纯 Java 库用%而不是%%引入别手滑写成%%否则 SBT 会去找 Scala 版本后缀的包直接报找不到。2.2 序列化器的选型逻辑Kafka 消息的 key 和 value 都需要序列化器。内置的有StringSerializer、ByteArraySerializer、IntegerSerializer等。但实际项目里消息体通常是复杂对象你得自己实现Serializer[T]和Deserializer[T]。这里有个关键决策序列化器要不要带 schema 信息。纯 JSON 序列化简单但下游解析时如果字段变了容易炸Avro 或 Protobuf 带 schema 演进能力但引入 Schema Registry 又增加运维成本。我的建议是内部系统、字段相对稳定的场景JSON 足够跨团队、字段频繁演进的场景老老实实上 Avro。下面是一个基于 Play JSON 的通用序列化器实现这个我用了好几年很稳import org.apache.kafka.common.serialization.{Deserializer, Serializer} import play.api.libs.json.{Format, Json} class JsonSerializer[T](implicit fmt: Format[T]) extends Serializer[T] { override def serialize(topic: String, data: T): Array[Byte] if (data null) null else Json.toJson(data).toString.getBytes(UTF-8) } class JsonDeserializer[T](implicit fmt: Format[T]) extends Deserializer[T] { override def deserialize(topic: String, data: Array[Byte]): T if (data null) null.asInstanceOf[T] else Json.parse(new String(data, UTF-8)).as[T] }注意serialize里对null的处理。Kafka 允许消息 value 为 null表示 tombstone用于日志压缩场景删除 key如果你的序列化器不处理 null遇到 tombstone 就会 NPE。这个坑我见过不止一个团队踩。3. 生产者实例从能跑到跑得稳3.1 最小可用生产者与参数逐条拆解先给一个能跑的最小生产者然后逐条讲参数import org.apache.kafka.clients.producer._ import org.apache.kafka.common.serialization.StringSerializer import java.util.Properties object SimpleProducer { def main(args: Array[String]): Unit { val props new Properties() props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092) props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName) props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName) props.put(ProducerConfig.ACKS_CONFIG, all) props.put(ProducerConfig.RETRIES_CONFIG, 3) props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true) val producer new KafkaProducer[String, String](props) try { val record new ProducerRecord[String, String](user-events, user-1, login) val metadata producer.send(record).get() println(s发送成功: partition${metadata.partition()}, offset${metadata.offset()}) } finally { producer.close() } } }acksall意味着 leader 要等所有 ISRin-sync replicas副本都写入才返回成功。这是不丢消息的前提代价是延迟略高。如果设成1leader 写完就返回leader 挂了但副本还没同步消息就丢了。设成0更是发出去就不管性能最高但基本等于赌博。retries3配合enable.idempotencetrue才是正确姿势。幂等生产者是 Kafka 0.11 引入的关键特性它给每条消息分配序列号broker 端去重保证重试不会产生重复消息。但幂等有前提acks必须是allretries必须大于 0max.in.flight.requests.per.connection不能超过 5。这几个条件不满足幂等自动失效而且 Kafka 不会报错只是静默降级——这就是我开头那个案例的根因。3.2 异步发送与回调的正确写法同步send().get()会阻塞吞吐上不去。生产环境基本都用异步加回调producer.send(record, new Callback { override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit { if (exception ! null) { // 这里千万别只打日志就完事 logger.error(s发送失败: topic${record.topic()}, exception) } else { logger.debug(soffset${metadata.offset()}) } } })用 Scala 可以写得更优雅把回调包成Futureimport scala.concurrent.{Future, Promise} def sendAsync[K, V](producer: KafkaProducer[K, V], record: ProducerRecord[K, V]): Future[RecordMetadata] { val promise Promise[RecordMetadata]() producer.send(record, (metadata: RecordMetadata, exception: Exception) { if (exception ! null) promise.failure(exception) else promise.success(metadata) }) promise.future }实操心得回调里绝对不要做耗时操作。Kafka 生产者的回调是在 sender 线程里执行的你在回调里查数据库、发 HTTP 请求会直接阻塞整个发送管道吞吐断崖式下跌。要处理失败消息正确做法是丢到一个内存队列由单独的线程消费。3.3 分区策略与消息顺序的取舍消息发到哪个 partition由partitioner.class决定。默认策略是有 key 就按 key 的 hash 取模没 key 就轮询2.4 之后改成了 sticky partitioner攒一批再换分区减少小批次。顺序性只在单个 partition 内保证。如果你的业务要求同一用户的事件严格有序就必须用 userId 做 key让同一用户的消息落到同一分区。但这样会带来数据倾斜风险——某个大 V 用户的消息量可能是普通用户的几万倍全压在一个分区上那个分区的 broker 就成了瓶颈。我处理过一个案例某活动期间头部几个用户的消息占了 40% 流量全挤在三个分区上消费端 lag 直接飙到百万。后来改成userId 时间窗口做 key牺牲了跨窗口的顺序性换来了负载均衡。所以顺序性和吞吐量往往要权衡没有银弹。4. 消费者实例提交策略决定一切4.1 消费者组与重平衡机制消费者最核心的概念是消费者组consumer group。同一个组内的消费者分摊分区一个分区只能被组内一个消费者消费。这个约束决定了消费并行度的上限就是分区数——你开 10 个消费者但 topic 只有 3 个分区那 7 个消费者就是白占资源。重平衡rebalance是消费者最头疼的事。触发条件有三个消费者加入、消费者离开、分区数变化。重平衡期间整个组会暂停消费eager 协议下如果频繁触发消费延迟会非常明显。Kafka 2.4 引入了增量协作重平衡CooperativeStickyAssignor重平衡时只迁移必要的分区其他消费者继续工作。这个改进很大新项目建议直接用props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, org.apache.kafka.clients.consumer.CooperativeStickyAssignor)4.2 手动提交 vs 自动提交一个必须想清楚的选择自动提交enable.auto.committrue看起来省事但有个致命问题它是在 poll 循环里按时间间隔提交的提交的是上次 poll 返回的 offset。如果你 poll 了一批消息处理到一半崩了自动提交可能已经把 offset 提交了那批没处理完的消息就永远丢了。手动提交才是生产环境的正确选择。手动提交又分两种提交方式时机风险适用场景commitSync处理完后同步提交阻塞吞吐低消息重要、不能丢commitAsync处理完后异步提交失败不重试可能重复吞吐优先、能容忍重复同步异步混合常规异步关闭前同步兼顾两者大多数生产场景混合提交是我最推荐的模式代码长这样val consumer new KafkaConsumer[String, String](props) consumer.subscribe(Collections.singletonList(user-events)) try { while (running) { val records consumer.poll(Duration.ofMillis(500)) records.forEach { record processRecord(record) // 业务处理 } if (!records.isEmpty) { consumer.commitAsync() // 常规异步提交 } } } finally { try { consumer.commitSync() // 关闭前同步提交确保不丢 } finally { consumer.close() } }commitAsync失败不重试是有意设计的——如果异步提交失败后重试可能提交的是更旧的 offset反而覆盖了更新的提交。所以常规用异步只在关闭时用同步兜底。4.3 消费位移与重复消费的本质很多人搞不清至少一次和最多一次到底怎么来的。其实就一句话先提交 offset 再处理消息就是最多一次可能丢先处理消息再提交 offset就是至少一次可能重复。Kafka 默认走的是至少一次。这意味着你的业务处理逻辑必须幂等。怎么做到幂等常见几种用消息里的唯一 ID 做去重表处理前先查用数据库的唯一索引重复插入直接失败用 Redis 的SETNX做短期去重窗口我一般推荐第一种虽然多一次查询但最可靠。去重表的清理策略要设计好比如保留 7 天用定时任务删过期数据。5. 完整可跑的实例一个端到端的消息管道5.1 消息模型定义把前面所有东西串起来做一个完整的例子模拟用户行为事件的生产和消费。import play.api.libs.json.{Format, Json} case class UserEvent( eventId: String, userId: String, eventType: String, timestamp: Long, payload: Map[String, String] ) object UserEvent { implicit val format: Format[UserEvent] Json.format[UserEvent] }eventId是幂等去重的关键生产端生成 UUID消费端用它查重。5.2 生产端完整实现import org.apache.kafka.clients.producer._ import java.util.Properties import java.util.UUID import scala.util.{Try, Success, Failure} class UserEventProducer(bootstrapServers: String) { private val props new Properties() props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers) props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName) props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[JsonSerializer[UserEvent]].getName) props.put(ProducerConfig.ACKS_CONFIG, all) props.put(ProducerConfig.RETRIES_CONFIG, 5) props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true) props.put(ProducerConfig.LINGER_MS_CONFIG, 10) props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768) props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, lz4) private val producer new KafkaProducer[String, UserEvent](props) def send(event: UserEvent): Try[RecordMetadata] { val record new ProducerRecord[String, UserEvent]( user-events, event.userId, event ) Try(producer.send(record).get()) } def close(): Unit producer.close() }linger.ms10和batch.size32768是吞吐优化的关键。生产者不会来一条发一条而是攒够batch.size或等够linger.ms才发一批。linger.ms从默认 0 调到 10吞吐能提升好几倍代价是最多 10ms 的额外延迟。这个取舍要看业务实时性要求极高的场景别调太大。compression.typelz4是性价比最高的压缩算法压缩比不错CPU 开销低。gzip 压缩比更高但慢snappy 快但压缩比一般。lz4 是大多数场景的甜点。5.3 消费端完整实现import org.apache.kafka.clients.consumer._ import java.util.{Collections, Properties} import java.time.Duration class UserEventConsumer(bootstrapServers: String, groupId: String) { private val props new Properties() props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers) props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId) props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getName) props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[JsonDeserializer[UserEvent]].getName) props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false) props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 200) props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000) props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, org.apache.kafka.clients.consumer.CooperativeStickyAssignor) private val consumer new KafkaConsumer[String, UserEvent](props) volatile private var running true def start(): Unit { consumer.subscribe(Collections.singletonList(user-events)) try { while (running) { val records consumer.poll(Duration.ofMillis(500)) records.forEach { record processWithRetry(record.value()) } if (!records.isEmpty) consumer.commitAsync() } } finally { try consumer.commitSync() finally consumer.close() } } private def processWithRetry(event: UserEvent): Unit { // 幂等处理先查 eventId 是否已处理 if (!isProcessed(event.eventId)) { doProcess(event) markProcessed(event.eventId) } } def shutdown(): Unit running false }max.poll.records200和max.poll.interval.ms300000这两个参数要配合着看。如果一批消息处理时间超过max.poll.interval.ms消费者会被踢出组触发重平衡。所以要么调大这个值要么调小max.poll.records让每批处理更快。我一般根据单条处理耗时反推单条 50ms一批 200 条就是 10 秒max.poll.interval.ms设 5 分钟绰绰有余。6. 常见问题排查与避坑清单6.1 生产端典型问题速查现象可能原因排查方向发送超时TimeoutExceptionbroker 不可达或网络抖动检查bootstrap.servers、防火墙、DNS消息重复幂等未生效检查 acks/retries/in.flight 三件套吞吐上不去linger.ms 太小、批次太小调大 linger 和 batch.sizeRecordTooLargeException单条消息超过max.request.size调大参数或拆分消息序列化异常序列化器与消息类型不匹配检查泛型和隐式 Format6.2 消费端典型问题速查现象可能原因排查方向频繁重平衡处理超时或心跳超时调大 max.poll.interval.ms、session.timeout.ms消费 lag 持续增长消费能力不足加消费者不超过分区数、优化处理逻辑重复消费提交前崩溃业务幂等 合理提交策略消息丢失自动提交 处理失败改手动提交分区分配不均默认分配器不均衡换 CooperativeStickyAssignor6.3 几个只有踩过才知道的坑第一个坑close()不调用会丢消息。生产者有缓冲区send()返回不代表消息发出去了。程序退出前不调close()缓冲区里的消息全丢。我见过有人用System.exit(0)直接退缓冲区消息一条没发出去。正确做法是注册 shutdown hook或者用 try-finally 保证 close。第二个坑消费者线程不安全。KafkaConsumer不是线程安全的多线程共享一个实例会抛ConcurrentModificationException。要么一个线程一个消费者要么用单线程 poll、多线程处理的模式但提交 offset 要回到 poll 线程做。第三个坑wakeup()是唯一安全的跨线程中断方式。想优雅停止消费者别用Thread.interrupt()用consumer.wakeup()它会让poll()抛WakeupException你在 catch 里处理完再 close。第四个坑topic 自动创建是个陷阱。auto.create.topics.enabletrue时生产者往不存在的 topic 发消息会自动创建但分区数和副本数用的是 broker 默认值往往不是你想要的。生产环境建议关掉自动创建topic 由运维统一管理。7. 性能调优的取舍逻辑7.1 吞吐、延迟、可靠性三角Kafka 调优本质是在吞吐、延迟、可靠性三者之间找平衡点不可能同时最优。追求吞吐调大batch.size、linger.ms、compression.type消费端调大fetch.min.bytes、max.poll.records。追求低延迟linger.ms0、batch.size调小消费端fetch.min.bytes1。追求可靠acksall、enable.idempotencetrue、手动提交、min.insync.replicas2。我的经验是先明确业务底线再优化其他维度。金融交易类可靠性是底线吞吐和延迟都得让路日志采集类吞吐是底线偶尔丢几条无所谓。7.2 一个真实的调优过程之前有个日志采集场景单机生产吞吐只有 5 万条/秒目标是 20 万。我按这个顺序调第一步linger.ms从 0 调到 20吞吐直接翻倍到 11 万。因为原来每条消息一个请求网络往返开销巨大。第二步batch.size从 16KB 调到 128KB吞吐到 15 万。批次大了压缩效率也上来了。第三步compression.type从 none 改成 lz4吞吐到 19 万。压缩减少了网络传输量。第四步buffer.memory从 32MB 调到 64MB吞吐稳定在 21 万。缓冲区不够时send()会阻塞调大后发送更顺畅。整个过程没动 broker 配置纯客户端调优就达标了。所以别一上来就怀疑 broker客户端参数往往才是瓶颈。8. 写在最后的一点个人体会用 Scala 写 Kafka 生产消费语言本身带来的便利是次要的真正重要的是对 Kafka 客户端机制的理解。我见过太多人把KafkaProducer当成一个简单的send方法用参数全默认出了问题就抓瞎。其实每一个参数背后都是 Kafka 设计者对某种场景的权衡你理解了权衡才知道自己的场景该往哪边靠。如果让我给刚上手的人一条建议那就是先把acks、enable.idempotence、enable.auto.commit这三个参数彻底搞明白。它们分别对应生产不丢、生产不重、消费不丢是整条链路可靠性的地基。地基打牢了再谈吞吐和延迟优化才有意义。至于 Scala 的那些语法糖用熟了自然就顺手。Future包异步发送、case class定义消息、模式匹配处理多类型消息这些写起来确实比 Java 舒服。但记住舒服归舒服底层还是那套 Java 客户端出问题还得回到那一层去找答案。