📚 课程目录第2章生产者进阶

生产者进阶：自定义分区器、消息确认机制（acks）、幂等性与事务、批量发送与压缩优化

生产者这块，很多同学觉得不就是 send 一下就完事了吗？其实不然。生产者的配置和策略，直接决定了你的数据能不能安全、高效地到达 Broker。我见过太多因为生产者配置不当导致的丢数据、性能瓶颈问题。今天咱们就把这几个核心点掰开揉碎了讲清楚。

1. 自定义分区器：数据该去哪，你说了算

Kafka 默认的分区策略，要么是 key 的 hash，要么是轮询。但实际业务中，这往往不够用。

举个例子。我之前做过一个订单系统，需要把同一个用户的订单都发到同一个分区，方便下游按用户维度做状态机处理。默认的 hash 策略虽然能保证相同 key 到同一分区，但一旦分区数变了，hash 结果就全乱了。这时候就需要自定义分区器。

核心思路：实现 org.apache.kafka.clients.producer.Partitioner 接口，重写 partition() 方法。

public class OrderIdPartitioner implements Partitioner {
    private static final int PARTITION_COUNT = 8;

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 假设 key 是 orderId，格式：ORD_20250101_001
        String orderId = (String) key;
        // 提取用户ID（这里简化处理）
        String userId = orderId.split("_")[2];
        // 按用户ID取模，保证同一用户到同一分区
        return Math.abs(userId.hashCode()) % PARTITION_COUNT;
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

配置时加上一行：

props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, OrderIdPartitioner.class.getName());

避坑指南：我曾经犯过一个错——在分区器里依赖了外部存储（Redis）来查路由表。结果生产环境 Redis 抖动，分区器直接超时，整个生产者都 hang 住了。记住：分区器必须是无状态的、纯内存计算，别搞网络调用。

2. 消息确认机制（acks）：安全与性能的博弈

acks 参数是生产者的灵魂配置。它决定了生产者要等多少个副本确认，才算发送成功。

acks 值	含义	可靠性	延迟	适用场景
`acks=0`	不等待任何确认	最低	最低	日志、监控等可丢数据
`acks=1`	等待 leader 确认	中等	中等	大部分业务场景
`acks=-1` 或 `all`	等待所有 ISR 副本确认	最高	最高	金融、订单等强一致性场景

我个人习惯，核心业务用 acks=all，配合 min.insync.replicas 一起用。比如副本因子是 3，设置 min.insync.replicas=2，这样即使一个副本挂了，也能保证数据写入了至少两个副本。

注意：acks=all 不代表数据不会丢。如果所有 ISR 副本都挂了，而 leader 又恰好没来得及把数据同步给其他副本……嗯，这种情况虽然概率极低，但理论上还是可能丢。所以关键业务还得配合幂等性和事务。

3. 幂等性与事务：从“至少一次”到“精确一次”

先说说幂等性。说白了就是：生产者重试发送同一条消息，Broker 不会重复存储。

开启方式很简单：

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

开启后，生产者会为每条消息分配一个 producerId 和 序列号。Broker 会根据序列号去重。我遇到过一个问题：开启幂等性后，生产者的吞吐量下降了。后来发现是 max.in.flight.requests.per.connection 默认被降到了 1（为了保证顺序）。如果你不需要严格顺序，可以调大这个值。

事务则更进一步。它保证一批消息要么全部成功，要么全部失败。比如你从一个 Topic 消费，处理后写入另一个 Topic，事务能保证这个过程是原子性的。

// 初始化事务
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("topic-a", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic-b", "key2", "value2"));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

我的经验：事务的开销不小。我曾在一次压测中发现，开启事务后吞吐量下降了 30% 左右。所以别滥用，只在真正需要“精确一次”语义的场景下用，比如支付、对账。

4. 批量发送与压缩优化：榨干网络带宽

Kafka 生产者的高性能，很大程度上归功于批量发送。它不会每来一条消息就发一次网络请求，而是攒一批再发。

关键参数就两个：

batch.size：批次大小，默认 16KB。不是越大越好，太大了会浪费内存，太小了又起不到批量效果。
linger.ms：等待时间，默认 0。设为 0 意味着有消息就立即发，不等待。想提高吞吐量，可以设成 5-10ms。

举个例子：

props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);    // 32KB
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);        // 等待10ms

压缩就更直接了。Kafka 支持 gzip、snappy、lz4、zstd 四种压缩算法。

算法	压缩比	CPU 开销	推荐场景
gzip	高	高	带宽瓶颈，CPU 充裕
snappy	中	低	通用场景，平衡之选
lz4	中	极低	追求低延迟
zstd	最高	中高	日志归档，存储成本敏感

我个人偏好 snappy，压缩比不错，CPU 开销也低。如果数据量特别大，比如每天 TB 级的日志，我会用 zstd，虽然 CPU 高一点，但能省下不少磁盘和网络成本。

一个小技巧：压缩是在生产者端进行的，Broker 和消费者端会自动解压。所以只要生产者配了压缩，下游完全不用改代码。我曾经帮一个团队排查问题，发现他们生产者和消费者都配了压缩，结果 CPU 翻倍了——重复压缩！记住：只在生产者端配一次就够了。

知识体系总览

下面这张图把生产者进阶的核心知识点串起来了，方便你对照理解：

好了，生产者进阶的核心内容就这些。记住一句话：没有银弹。每个配置项都有 trade-off，关键是你得清楚自己的业务到底要什么——是要不丢数据，还是要极致吞吐量？想清楚了，配置起来就顺手了。