Kafka producer：acks、幂等性和批处理

1. 概述

向 Kafka 发送一条消息只需一行 send()。难的是接踵而至的问题：消息是否真的被持久化了？当 producer 重试时会不会被写入两次？如何每秒发送数万条消息而不让 broker 过载？

本文介绍三组配置来回答这三个问题：acks、enable.idempotence 和 batching（batch.size、linger.ms、compression.type）。读完这篇文章后，你将知道如何既安全又快速地配置一个 producer，无论是用原生 Kafka 客户端还是 Spring Boot 3.x，以及如何测试它。

2. 设置

使用原生 Java，你只需要 kafka-clients：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.9.0</version>
</dependency>

使用 Spring Boot 3.x，使用 spring-kafka。版本由 Spring Boot 管理，它已经包含了 kafka-clients：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

在整篇文章中，我们发送订单事件：

public record Order(String id, String customerId, BigDecimal total) {

    public String toJson() {
        return """
            {"id":"%s","customerId":"%s","total":%s}""".formatted(id, customerId, total);
    }
}

3. acks：什么时候认为消息已发送？

acks 规定了 producer 在认为一个请求成功之前必须等待来自 broker 的多少个确认。

值	含义	风险
acks=0	根本不等待	消息丢失而你毫不知情
acks=1	等待 leader 写完	如果 leader 在 follower 复制完成前宕机，消息会丢失
acks=all (-1)	等待 ISR 中的所有副本写完	延迟略高

使用原生客户端的配置示例：

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, "all");

try (var producer = new KafkaProducer<String, String>(props)) {
    var order = new Order("o-1001", "c-42", new BigDecimal("250000"));
    RecordMetadata meta = producer
        .send(new ProducerRecord<>("orders", order.customerId(), order.toJson()))
        .get();
    System.out.printf("partition=%d offset=%d%n", meta.partition(), meta.offset());
}

调用 .get() 使发送变成同步的。这里这样做只是为了打印结果。当 acks=all 时，.get() 只在 ISR 中的所有副本都写完后才返回：

partition=2 offset=17

我们使用 customerId 作为 key，所以同一个客户的所有订单都进入同一个 partition 并保持顺序。

3.1. 没有 min.insync.replicas 的话 acks=all 还不够

acks=all 意思是“当前在 ISR 中的所有副本”。假设 ISR 缩小到只有 leader。那么 acks=all 和 acks=1 就没区别了。你需要在 topic 或 broker 级别配置：

kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders \
  --partitions 6 --replication-factor 3 \
  --config min.insync.replicas=2

使用 replication-factor=3 和 min.insync.replicas=2，每次写入必须至少有 2 个副本。如果 ISR 降到 2 以下，producer 会收到 NotEnoughReplicasException 而不是悄悄地写入单个副本。

注意： 不要将 min.insync.replicas 设置等于 replication-factor。这样只需一个 broker 重启就会导致 topic 拒绝所有写入。

4. 幂等性：重试不重复

acks=all 需要重试。考虑这样一个场景：broker 已经写入消息但响应在路上丢失了。Producer 认为发送失败于是重新发送，结果消息被写入了两次。

幂等性 producer 解决了这个确切的场景。当启用 enable.idempotence=true：

• Broker 为 producer 分配一个 producer ID (PID)。
• 发送到某个 partition 的每个 batch 携带一个递增的 sequence number。
• Broker 丢弃已见过的 sequence，拒绝跳跃的 sequence。这样即使有多个请求在途中也能保持顺序。

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

每行的含义：

• ENABLE_IDEMPOTENCE_CONFIG 启用 PID 和 sequence number 机制。
• 幂等性强制要求 acks=all。如果你显式地将 acks=1 与 enable.idempotence=true 放在一起，producer 在初始化时会抛出 ConfigException。
• max.in.flight.requests.per.connection 必须 ≤ 5。在此限制内，Kafka 仍然保证 partition 内的顺序。
• retries 设置得很大。实际重试时间受 delivery.timeout.ms 限制（默认 2 分钟）。

从 Kafka 3.0 开始，acks=all 和 enable.idempotence=true 已经是 默认值。尽管如此，你应该明确声明。否则，只要有人设置 acks=1，幂等性就会默默关闭。

4.1. 幂等性做不到的事

这是容易被误解的部分：

• 它只防止在 单个 producer 会话内、按 partition 重复。如果应用重启，producer 会获得新的 PID。重启后发送的消息仍然会重复。
• 它不防止由于 你的代码 调用两次 send() 造成的重复，例如当你在 catch 中自己重试时。
• 它不能对多个 partition 进行原子性写入。

这些情况需要 transaction（transactional.id）或能够按 orderId 去重的 consumer。

5. 批处理：用延迟换吞吐量

Producer 不是逐条发送消息。send() 只是将记录放入内存中的缓冲区，按 partition 组织。一个后台线程（sender）将多条记录组合成 batch 然后发送。三个主要参数：

参数	默认值	作用
batch.size	16384 字节	单个 partition 的 batch 最大大小
linger.ms	0（Kafka 4.0 后为 5）	等待更多数据来填满 batch 的时间
compression.type	none	压缩整个 batch：gzip、snappy、lz4、zstd
buffer.memory	32 MB	所有待发送 batch 的总内存

当 batch 达到 batch.size 或 linger.ms 时间过去 时发送，以先发生者为准。使用 linger.ms=0 时，batch 几乎总是立即发送，通常只包含几条记录。

props.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

这个配置最多增加约 10 ms 的延迟。作为交换，每个请求携带的是 64 KB 已压缩的数据。对于订单事件这样有大量重复的 JSON 数据，lz4 或 zstd 通常能显著减少网络带宽和 broker 的磁盘使用。按 batch 压缩比单条消息压缩效率高得多，因为 batch 越大重复的串就越多。

你可以通过 producer 的 metric 来验证：

producer.metrics().forEach((name, metric) -> {
    if (name.name().equals("batch-size-avg") || name.name().equals("records-per-request-avg")) {
        System.out.println(name.name() + " = " + metric.metricValue());
    }
});

连续发送 10,000 个订单，先用 linger.ms=0，再用 linger.ms=10，你会看到 records-per-request-avg 显著增加：

# linger.ms=0
batch-size-avg = 1830.4
records-per-request-avg = 21.7
# linger.ms=10
batch-size-avg = 41022.9
records-per-request-avg = 486.2

具体数值取决于机器和负载，但趋势总是这样。

注意： 如果缓冲区满了（与 consumer 速度无关，这里是 broker 太慢或网络拥塞），send() 会 阻塞 最多 max.block.ms（默认 60 秒）然后抛出 TimeoutException。不要在处理 HTTP 请求的线程中调用 send() 而不考虑这种可能性。

注意： delivery.timeout.ms 必须 ≥ linger.ms + request.timeout.ms。否则 producer 会拒绝初始化。

6. Spring Boot 3.x 中的配置

Spring Boot 为常见参数提供了自己的属性。其他参数通过 properties 传递：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      batch-size: 64KB
      compression-type: lz4
      buffer-memory: 64MB
      properties:
        enable.idempotence: true
        linger.ms: 10
        max.in.flight.requests.per.connection: 5
        delivery.timeout.ms: 120000

Publisher 使用 KafkaTemplate。从 Spring Kafka 3.0 开始，send() 返回 CompletableFuture：

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, String>> publish(Order order) {
        return kafkaTemplate.send("orders", order.customerId(), order.toJson())
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("发送订单 {} 失败", order.id(), ex);
                } else {
                    var meta = result.getRecordMetadata();
                    log.info("订单 {} -> partition {} offset {}",
                        order.id(), meta.partition(), meta.offset());
                }
            });
    }
}

注意三点：

• Callback 在 producer 的 sender 线程 上运行。不要在 whenComplete 中做重操作或调用阻塞方法，因为会拖累所有其他发送。
• 当 callback 收到异常时，producer 已经在 delivery.timeout.ms 限制内重试过了。不要再手动重试，比如立即再次调用 send()。这正是幂等性无法阻止的重复来源。
• 返回 CompletableFuture 让调用者自己决定是否等待。

7. 测试 producer 行为

kafka-clients 提供 MockProducer 用于无 broker 的测试。要使用它，我们需要将逻辑提取到接受 Producer 接口的类中：

public class OrderSender {

    private final Producer<String, String> producer;

    public OrderSender(Producer<String, String> producer) {
        this.producer = producer;
    }

    public Future<RecordMetadata> send(Order order) {
        return producer.send(new ProducerRecord<>("orders", order.customerId(), order.toJson()));
    }

    public static Properties reliableConfig(String bootstrap) {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
        p.put(ProducerConfig.ACKS_CONFIG, "all");
        p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        p.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        p.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        p.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
        p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        return p;
    }
}

使用 JUnit 5 和 AssertJ 的测试：

class OrderSenderTest {

    @Test
    void whenSendingOrders_thenKeyIsCustomerId() {
        var mock = new MockProducer<>(true, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        sender.send(new Order("o-1", "c-42", new BigDecimal("100")));
        sender.send(new Order("o-2", "c-42", new BigDecimal("200")));

        assertThat(mock.history())
            .hasSize(2)
            .extracting(ProducerRecord::key)
            .containsOnly("c-42");
    }

    @Test
    void whenBrokerFails_thenFutureCompletesExceptionally() {
        var mock = new MockProducer<>(false, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        Future<RecordMetadata> future = sender.send(new Order("o-3", "c-7", BigDecimal.TEN));
        mock.errorNext(new NotEnoughReplicasException("ISR < min.insync.replicas"));

        assertThatThrownBy(future::get)
            .hasCauseInstanceOf(NotEnoughReplicasException.class);
    }

    @Test
    void reliableConfig_isAccepted() {
        var props = OrderSender.reliableConfig("localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        var config = new ProducerConfig(props);

        assertThat(config.getString(ProducerConfig.ACKS_CONFIG)).isEqualTo("-1");
        assertThat(config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)).isTrue();
    }
}

每个测试检查一个行为：

• 第一个测试 使用 autoComplete=true，所以每个 send() 都立即成功。history() 允许检查 key，从而确认按客户顺序保持。
• 第二个测试 使用 autoComplete=false 然后 errorNext(...) 来模拟 broker 拒绝写入。这是检查你的错误处理代码的方法。
• 最后的测试 初始化 ProducerConfig 让 Kafka 自己验证配置组合。错误的组合，如 acks=1 加上幂等性，会导致测试用 ConfigException 失败而不是在部署时爆炸。Kafka 将 all 标准化为 "-1"，所以 assertion 比较 "-1"。

8. 何时应该放松配置

“最大安全”配置不是总正确的：

• Metrics、日志、clickstream：丢失几条记录可以接受。此时 acks=1、高 linger.ms 和 zstd 压缩能提供最好吞吐量。
• 需要低单消息延迟的系统（例如实时报价）：保持 linger.ms 小（0-2 ms）。这种情况下不要期望大 batch。
• 支付、订单：总是使用 acks=all、幂等性和 min.insync.replicas=2。如果需要“读-处理-写”精确一次，考虑 transaction。

注意： 设置 max.in.flight.requests.per.connection=1 来“保险”是不必要的，当已经启用幂等性时。这样只会降低吞吐量。

9. 结论

三组配置解决三个不同的问题：

• acks=all 结合 min.insync.replicas 确保 不丢失 消息。
• enable.idempotence 确保重试 不重复 并保持 partition 内的顺序。它的范围仅限于单个 producer 会话。
• batch.size、linger.ms 和 compression.type 决定 吞吐量，代价是一些延迟。

应该明确声明这些值，而不是依赖默认值。通过测试验证它们以避免部署时的惊喜。不要在 producer 已经为你做的事情上自己重试 send()。