4. 事件生产与消费编程模型

Kafka 的客户端编程模型围绕两个核心抽象展开：Producer 负责把事件写入主题分区，Consumer 负责从分区拉取事件并推进消费位移。理解这两个 API 的参数语义、序列化机制、位移提交策略以及幂等生产者的配置方式，是构建可靠事件驱动系统的起点。本章不讨论集群运维，而是聚焦客户端代码层面的关键决策点。

Producer API 的核心参数

创建一个 KafkaProducer 时，最关键的配置项决定了吞吐、延迟与可靠性之间的权衡。以下是一段典型的 Java 配置示例：

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("acks", "all");
props.put("retries", 5);
props.put("enable.idempotence", true);
props.put("linger.ms", 20);
props.put("batch.size", 32768);
props.put("max.in.flight.requests.per.connection", 5);
KafkaProducer<String, OrderEvent> producer = new KafkaProducer<>(props);

acks 控制 broker 在什么条件下向生产者确认写入成功。acks=0 表示不等待确认，吞吐最高但可能丢消息；acks=1 表示 leader 写入本地日志即确认；acks=all（或 -1）表示 ISR 中所有副本都确认，可靠性最高。生产环境通常选择 acks=all。

retries 决定可重试错误的次数。配合 delivery.timeout.ms（默认 120000 毫秒）共同约束重试总时长。需要注意的是，开启重试后如果未启用幂等性，可能因重试导致消息重复。

linger.ms 与 batch.size 共同影响批量发送行为。linger.ms 让生产者在发送前等待一小段时间以积累更多消息，batch.size 则限制单个批次的最大字节数。两者配合可以在吞吐与延迟之间取得平衡。示例中 linger.ms=20 表示最多等待 20 毫秒，batch.size=32768 表示每批最多 32KB。

max.in.flight.requests.per.connection 限制每个连接上未确认请求的数量。当启用幂等性时，该值最大为 5，Kafka 通过序列号保证即使存在多个在途请求也不会乱序。

序列化与反序列化

Kafka 只传输字节数组，因此生产者和消费者必须约定序列化格式。常见的序列化器包括 StringSerializer、ByteArraySerializer、IntegerSerializer 等基础类型，以及 Confluent 提供的 Avro、Protobuf、JSON Schema 序列化器。

使用 Schema Registry 时，序列化器会在消息体中嵌入 schema ID。生产者发送时，KafkaAvroSerializer 先向 Schema Registry 注册或查询 schema，获取全局唯一的 schema ID，然后按照“魔术字节 + schema ID + Avro 二进制”的格式写入。消费者端 KafkaAvroDeserializer 读取 schema ID 后从 Registry 拉取对应 schema 完成反序列化。这种机制使得 schema 演进成为可能，同时保持向后兼容。

如果使用自定义序列化器，需要实现 Serializer<T> 接口的 serialize 方法，并确保线程安全。反序列化器则实现 Deserializer<T> 接口的 deserialize 方法。一个常见的错误是生产者和消费者使用了不匹配的序列化格式，导致消费端抛出 SerializationException。

消费位移提交策略

消费者通过 group.id 加入消费组，Kafka 为每个分区分配一个消费者实例。消费者需要定期提交位移（offset），记录自己消费到了哪里。位移提交策略直接影响消息投递语义。

自动提交是最简单的方式。设置 enable.auto.commit=true 后，消费者在 poll 循环中每隔 auto.commit.interval.ms（默认 5000 毫秒）自动提交一次当前位移。问题在于：如果消费者拉取了消息但处理失败，位移可能已经被提交，导致消息丢失。因此自动提交适合对消息丢失不敏感的场景。

同步提交通过 consumer.commitSync() 显式提交，调用会阻塞直到 broker 确认。它提供了最强的控制力，但每次提交都增加延迟。通常的做法是处理完一批消息后调用一次 commitSync()。

异步提交通过 consumer.commitAsync() 提交，不阻塞后续 poll。可以传入 OffsetCommitCallback 处理提交结果。异步提交的典型问题是：如果提交失败且没有重试，消费者重启后会从旧位移开始，导致重复消费。实践中常将 commitAsync 与 commitSync 结合使用——正常循环用异步，关闭前用同步做最终提交。

另一个关键参数是 auto.offset.reset，它决定当消费者找不到已提交位移或位移越界时的行为。earliest 表示从最早的消息开始消费，latest 表示从最新消息开始，none 则抛出异常。新消费组首次启动时，该参数尤为重要。

幂等生产者配置

幂等生产者是 Kafka 实现精确一次语义的基础。启用方式非常简单：设置 enable.idempotence=true。Kafka 会自动调整相关参数——acks 被强制为 all，retries 至少为 Integer.MAX_VALUE，max.in.flight.requests.per.connection 不超过 5。

其工作原理是：生产者启动时向 broker 申请一个生产者 ID（PID），broker 为每个 PID 维护每个分区的序列号。生产者发送的每条消息都携带 (PID, 分区, 序列号)。broker 只接受序列号恰好比当前值大 1 的消息，重复序列号会被丢弃。这样即使发生网络重试，broker 也能识别并去重。

需要注意的是，幂等生产者只保证单个生产者会话内、单个分区上的精确一次。如果生产者重启，PID 会变化，跨会话的去重需要配合事务（transactional.id）来实现。此外，幂等性不保证跨分区的原子性，那需要事务 API 的支持。

参数协同与常见陷阱

Producer 和 Consumer 的参数并非孤立存在。例如，Producer 的 linger.ms 和 batch.size 影响消息到达 broker 的延迟，进而影响 Consumer 的 poll 返回时机。Consumer 的 max.poll.records 控制单次 poll 返回的最大记录数，如果设置过大而处理逻辑耗时较长，可能触发 max.poll.interval.ms（默认 300000 毫秒）超时，导致消费者被踢出组并触发再平衡。

另一个常见陷阱是序列化器不匹配。Producer 使用 Avro 序列化而 Consumer 使用 String 反序列化，会在消费端抛出 SerializationException。解决方法是确保两端使用相同的序列化格式和 Schema Registry 地址。

最后，位移提交的时机与处理逻辑的幂等性需要配合设计。即使使用了幂等生产者，消费端如果重复处理消息（例如异步提交失败后重试），仍然可能产生重复副作用。因此，消费端处理逻辑最好也设计为幂等的，例如使用唯一键去重或 upsert 操作。