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() } } }

acks=all意味着 leader 要等所有 ISR（in-sync replicas）副本都写入才返回成功。这是不丢消息的前提，代价是延迟略高。如果设成1，leader 写完就返回，leader 挂了但副本还没同步，消息就丢了。设成0更是发出去就不管，性能最高但基本等于赌博。

retries=3配合enable.idempotence=true才是正确姿势。幂等生产者是 Kafka 0.11 引入的关键特性，它给每条消息分配序列号，broker 端去重，保证重试不会产生重复消息。但幂等有前提：acks必须是all，retries必须大于 0，max.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(s"offset=${metadata.offset()}") } } })

用 Scala 可以写得更优雅，把回调包成Future：

import 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.commit=true）看起来省事，但有个致命问题：它是在 poll 循环里按时间间隔提交的，提交的是"上次 poll 返回的 offset"。如果你 poll 了一批消息，处理到一半崩了，自动提交可能已经把 offset 提交了，那批没处理完的消息就永远丢了。

手动提交才是生产环境的正确选择。手动提交又分两种：

混合提交是我最推荐的模式，代码长这样：

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.ms=10和batch.size=32768是吞吐优化的关键。生产者不会来一条发一条，而是攒够batch.size或等够linger.ms才发一批。linger.ms从默认 0 调到 10，吞吐能提升好几倍，代价是最多 10ms 的额外延迟。这个取舍要看业务，实时性要求极高的场景别调太大。

compression.type=lz4是性价比最高的压缩算法，压缩比不错，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.records=200和max.poll.interval.ms=300000这两个参数要配合着看。如果一批消息处理时间超过max.poll.interval.ms，消费者会被踢出组，触发重平衡。所以要么调大这个值，要么调小max.poll.records让每批处理更快。我一般根据单条处理耗时反推：单条 50ms，一批 200 条就是 10 秒，max.poll.interval.ms设 5 分钟绰绰有余。

6. 常见问题排查与避坑清单

6.1 生产端典型问题速查

6.2 消费端典型问题速查

6.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.enable=true时，生产者往不存在的 topic 发消息会自动创建，但分区数和副本数用的是 broker 默认值，往往不是你想要的。生产环境建议关掉自动创建，topic 由运维统一管理。

7. 性能调优的取舍逻辑

7.1 吞吐、延迟、可靠性三角

Kafka 调优本质是在吞吐、延迟、可靠性三者之间找平衡点，不可能同时最优。

追求吞吐：调大batch.size、linger.ms、compression.type，消费端调大fetch.min.bytes、max.poll.records。

追求低延迟：linger.ms=0、batch.size调小，消费端fetch.min.bytes=1。

追求可靠：acks=all、enable.idempotence=true、手动提交、min.insync.replicas>=2。

我的经验是，先明确业务底线，再优化其他维度。金融交易类，可靠性是底线，吞吐和延迟都得让路；日志采集类，吞吐是底线，偶尔丢几条无所谓。

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 客户端，出问题还得回到那一层去找答案。