Kafka Producer：正确处理 acks、retry 和幂等性

1. 概览

当 producer 向 Kafka 发送消息时，你需要回答三个问题：发送完成后是否应该被视为成功、出错时如何重试、重试是否会导致消息重复。本文解释 acks、retries 和 enable.idempotence —— 三个配置项决定了 producer 的可靠性和正确性。阅读后，你将知道如何配置“生产就绪”的 producer，并避免常见的数据丢失或重复问题。

2. 必需的依赖项

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

本文使用纯 Kafka client（KafkaProducer），不经过 Spring Kafka，以便清楚地看到每个配置如何产生影响。

3. acks：producer 何时认为发送成功

acks 决定了 producer 需要等待多少个 broker 的确认才能认为消息已发送完成。

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

• acks=0：producer 不等待任何确认。速度最快，但如果 broker 在接收后立即宕机，消息可能会丢失——甚至不知道已丢失。
• acks=1：只等待 leader 分区将消息写入本地日志后返回 ACK。如果 leader 在 follower 同步之前崩溃，消息会丢失。
• acks=all（或 -1）：等待 ISR（同步副本集）中的所有副本确认。最安全，但延迟更高。

注意： acks=all 只有在配合 broker 上的 min.insync.replicas >= 2 时才真正安全。如果 min.insync.replicas=1，acks=all 也只需要 1 个副本的确认——在安全性上与 acks=1 没有区别。

3.1. 配套的 broker 配置

# server.properties 或 topic 配置
min.insync.replicas=2

当 replication factor = 3 且 min.insync.replicas=2 时，系统可以容忍 1 个 broker 宕机仍能继续写入，同时保证数据在返回 ACK 前至少位于 2 个副本上。

4. retries：处理临时错误

Producer 可能遇到诸如 NotLeaderForPartitionException 或短暂网络连接丢失等临时错误。retries 允许 producer 自动重新发送，无需手动处理代码。

props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

• retries：最大重试次数。从 Kafka 2.1 开始，应该设置很高（或 Integer.MAX_VALUE），并使用 delivery.timeout.ms 来控制总等待时间，而不是计算重试次数。
• retry.backoff.ms：重试之间的等待时间，防止 spam broker。
• delivery.timeout.ms：从调用 send() 到收到结果（成功或最终错误）的最大总时间。必须大于 request.timeout.ms + linger.ms。

问题： 如果不小心，重试可能导致消息发送重复或发送顺序错误。

4.1. 为什么重试会导致消息重复

1. Producer 发送包含消息 A 的批次
2. Broker 成功写入 A，发送 ACK
3. ACK 在网络上丢失（网络超时）
4. Producer 没有收到 ACK → 认为出错 → 重试
5. Broker 再次收到请求 → 再次写入 A
=> Topic 中有 2 份消息 A

这是一个经典情况：broker 已成功写入，但 producer 不知道，所以重新发送。后面的幂等性部分解决了这个问题。

4.2. 为什么重试会导致顺序错误

如果 max.in.flight.requests.per.connection > 1，producer 可以并行发送多个批次。如果批次 1 出错并正在重试，批次 2 可能会在批次 1 之前被 broker 写入。

max.in.flight.requests.per.connection=5

在这种配置下，如果没有幂等性，发生重试时，partition 中消息的顺序无法保证。

5. 幂等性：解决重试导致的重复

enable.idempotence=true 帮助 producer 在重试时自动避免重复写入，这是通过 Kafka 为每个批次分配 Producer ID (PID) 和 序列号 实现的。

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

启用幂等性时，Kafka 会自动设置以下值（你不需要手动配置，但应该了解）：

acks = all
retries = Integer.MAX_VALUE （或很大的值）
max.in.flight.requests.per.connection <= 5

5.1. 工作机制

每个 producer session 被 broker 分配一个唯一的 PID。每条在 partition 中发送的消息都有一个递增的 序列号。Broker 为每个 PID + partition 组合保存最后写入的序列号。

Producer 向 partition 0 发送序列号为 5 的批次
Broker："我保存的最后序列号是 4，5 是下一个 → 写入，保存 seq=5"
Producer 重新发送（由于重试）序列号为 5 的批次
Broker："我已经写入过 seq=5 了 → 跳过，直接返回成功 ACK"

这样，重试不会创建重复记录——broker 根据序列号自动检测和排除重复，无需 producer 或 consumer 手动处理。

注意： 幂等性只保证在一个 producer session 内和在一个 partition 范围内不重复。如果 producer 重启（分配新的 PID），基于旧 PID 的 dedup 机制不再适用。如果你想在多个 partition/topic 上实现完全一次性语义，你需要事务（transactional.id）——这超出了单纯幂等性的范围。

5.2. 生产就绪的 producer 配置示例

public ProducerFactory() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // 启用幂等性 -> 自动设置 acks=all、重试次数高、in-flight <= 5
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

    this.producer = new KafkaProducer<>(props);
}

public void sendOrderEvent(String orderId, String payload) {
    ProducerRecord<String, String> record =
            new ProducerRecord<>("order-events", orderId, payload);

    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            log.error("发送订单 {} 失败", orderId, exception);
        } else {
            log.info("订单 {} 已写入 partition {} offset {}",
                    orderId, metadata.partition(), metadata.offset());
        }
    });
}

• ENABLE_IDEMPOTENCE_CONFIG = true 是最重要的一行——它自动启用 acks=all。
• send() 中的回调函数告诉你实际结果：如果 exception != null，意味着在重试完 delivery.timeout.ms 后仍然出错——此时你才需要处理业务逻辑（日志、告警、保存到错误表）。

6. 通过测试验证行为

为了演示 dedup 部分，我们模拟一个简单的 MessageDeduplicator，在应用程序级别模拟序列号机制（在你无法启用幂等性的情况下使用，例如遗留系统）：

public class MessageDeduplicator {

    private final Map<String, Long> lastSeenSequence = new ConcurrentHashMap<>();

    public boolean shouldWrite(String partitionKey, long sequence) {
        Long last = lastSeenSequence.get(partitionKey);
        if (last != null && sequence <= last) {
            return false; // 重复，跳过
        }
        lastSeenSequence.put(partitionKey, sequence);
        return true;
    }
}

class MessageDeduplicatorTest {

    @Test
    void shouldSkipDuplicateSequence() {
        MessageDeduplicator deduplicator = new MessageDeduplicator();

        assertThat(deduplicator.shouldWrite("order-1", 5)).isTrue();
        assertThat(deduplicator.shouldWrite("order-1", 5)).isFalse(); // 重试相同序列号
        assertThat(deduplicator.shouldWrite("order-1", 6)).isTrue();
    }

    @Test
    void shouldAllowDifferentPartitionKeysIndependently() {
        MessageDeduplicator deduplicator = new MessageDeduplicator();

        assertThat(deduplicator.shouldWrite("order-1", 1)).isTrue();
        assertThat(deduplicator.shouldWrite("order-2", 1)).isTrue();
    }
}

第一个测试断言：相同的序列号第二次发送时被跳过——符合 producer 幂等性的精神。第二个测试断言：dedup 仅适用于一个 key/partition 范围，不影响其他 key。

7. 根据需求总结配置

需求	acks	enable.idempotence	风险
日志、指标不重要	0	false	出错时丢失消息，但吞吐量最高
速度/安全性的平衡	1	false	如果 leader 在 follower 同步前崩溃，消息会丢失
重要业务数据（订单、支付）	all	true	延迟更高，但不丢失、不会因重试而重复

8. 常见错误

• 设置 acks=all 但忘记 min.insync.replicas：以为很安全，但 broker 只需 1 个 ISR 就会 ACK——与 acks=1 没有区别。
• 与 Kafka client 默认重试逻辑并行自己编写重试逻辑：容易导致消息发送重复加倍，因为两层重试都在运行。
• 启用幂等性但设置 max.in.flight.requests.per.connection > 5：Kafka 会在 producer 初始化时立即拒绝此配置（ConfigException），因为超过了幂等性支持的限制。
• 混淆幂等性与系统级别的完全一次性语义：幂等性只保护 producer-to-broker。如果 consumer 处理消息后调用非幂等的副作用（发送邮件、扣款），你仍然可能在 consumer 端处理重复。

注意： 如果系统运行的是早于 0.11 的 Kafka 版本，不应该启用 enable.idempotence=true——broker 还不支持此功能，producer 初始化时会出错。

9. 结论

acks、retries 和 enable.idempotence 是你需要可靠 producer 时三个不可分割的拼图。acks=all 确保数据在被视为成功之前被充分复制；retries 帮助克服临时错误；幂等性确保重试不会创建重复记录。对于重要业务数据，默认应该同时使用这三者——延迟成本远小于事后处理错误或重复数据的成本。