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

生产者进阶：分区策略、自定义Partitioner、消息键与顺序保证、幂等性与事务

说实话，很多同学用Kafka生产消息，就是调个send()完事。但线上出问题的时候，往往就卡在分区策略和顺序保证上。我见过太多因为分区没设计好，导致数据倾斜、消费延迟飙升的案例了。

这一节，咱们把生产者的几个进阶能力掰开揉碎讲清楚。你想想看，消息到底落到哪个分区？怎么保证同一个订单的消息按顺序被消费？幂等和事务到底解决了什么问题？

分区策略：消息到底去了哪？

Kafka的一个Topic可以分成多个Partition。消息发到哪个分区，不是随机的。默认情况下，分区策略由Partitioner接口的实现类决定。

Kafka内置了几种分区策略：

策略	说明	适用场景
DefaultPartitioner	有key时按key哈希，无key时轮询	通用场景
RoundRobinPartitioner	无key时严格轮询	负载均衡要求高
UniformStickyPartitioner	粘性分区，减少网络开销	高吞吐批量发送

我个人习惯，只要业务上需要按某个字段聚合，就一定指定消息键。比如订单ID、用户ID。为什么呢？因为同一个key的消息，默认会落到同一个分区。这是顺序保证的基础。

核心原则：有key走哈希，无key走轮询。但轮询在2.4版本之后默认变成了粘性分区，说白了就是攒一批再发，减少请求次数。

自定义Partitioner：把控制权握在自己手里

内置策略不够用怎么办？比如我想让某个地域的数据单独进一个分区，或者按流量比例分配。这时候就得自己写Partitioner。

我在项目中遇到过这样一个场景：用户行为数据，北京、上海的用户占了60%，其他地区占40%。如果按用户ID哈希，分区数据量会严重不均。我写了一个自定义Partitioner，按地域编码分配，热点地区多分几个分区。

实现起来很简单：

public class RegionPartitioner implements Partitioner {
    private static final Map<String, Integer> REGION_MAP = new HashMap<>();
    static {
        REGION_MAP.put("BJ", 0);
        REGION_MAP.put("SH", 1);
        REGION_MAP.put("OTHER", 2);
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        String region = extractRegion(key.toString());
        return REGION_MAP.getOrDefault(region, 2);
    }

    @Override
    public void close() {}

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

然后在生产者配置里指定：

props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RegionPartitioner.class);

小提示：自定义Partitioner里不要做耗时操作，比如查数据库、调远程接口。它是在发送线程里同步调用的，阻塞了会影响整个生产者的吞吐量。

消息键与顺序保证

顺序保证是Kafka里最容易踩坑的地方。很多人以为Kafka能保证全局顺序，其实不是。Kafka只保证同一个分区内的消息是有序的。

举个例子：订单状态流转，从"已创建"到"已支付"到"已发货"。如果这三个消息落到不同分区，消费端拿到的顺序可能是乱的。

解决办法很简单——用订单ID作为消息key。同一个订单的消息永远进同一个分区，消费端就能按发送顺序处理。

我曾经踩过的坑：有次做订单系统，key用了用户ID。结果同一个用户的多个订单混在一个分区里，订单A和订单B的消息交叉到达，消费端处理逻辑直接崩了。后来改成订单ID做key，问题解决。记住：顺序保证的粒度取决于key的粒度。

另外要注意，分区数变了怎么办？比如从3个扩展到6个。同一个key可能被重新哈希到新分区。这时候顺序保证就断了。所以生产环境的分区数，我建议一开始就规划好，尽量不扩容。

幂等性：消息不重不漏

网络抖动、Broker宕机，都可能导致生产者重试发送。重试意味着同一条消息可能被写入多次。幂等性就是解决这个问题的。

开启方式很简单：

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

底层原理是Kafka给每个生产者分配了一个Producer ID（PID），每条消息带一个序列号。Broker会检查序列号，重复的就直接丢弃。

但注意，幂等性只保证单次会话内的不重复。如果生产者挂了重启，PID变了，之前的序列号就失效了。跨会话的精确一次，需要事务来保证。

特性	幂等性	事务
范围	单分区单会话	跨分区跨会话
配置	enable.idempotence=true	transactional.id + 幂等
开销	低	较高
适用场景	大部分业务	金融、对账等

我的建议：只要不是对性能极度敏感的场景，都开启幂等性。它几乎不增加延迟，但能省掉你排查重复数据的痛苦。

事务：跨分区原子写入

幂等性搞不定跨分区、跨Topic的原子性。比如从订单Topic读一条消息，处理后写入支付Topic和库存Topic。如果写支付成功但写库存失败，数据就不一致了。

Kafka事务就是干这个的。它保证一批消息要么全部写入成功，要么全部不可见。

使用事务的步骤：

配置transactional.id，每个生产者实例唯一
调用initTransactions()初始化
用beginTransaction()开启事务
正常发送消息
调用commitTransaction()或abortTransaction()

props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-txn-001");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("payment", orderId, paymentInfo));
    producer.send(new ProducerRecord<>("inventory", orderId, inventoryInfo));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

注意：事务不是银弹。它增加了延迟和Broker负载。我一般只在跨系统对账、金融交易这类场景才用。普通业务，幂等性加手动补偿就够了。

知识体系总览

下面这张图，把生产者进阶的核心逻辑串起来了：

从图上能看出来，消息从生产者发出，先经过分区策略决定目标分区。如果带了消息键，同一个key永远走同一个分区，顺序就有了保障。然后根据业务需求，选择幂等性还是事务来保证不重复或原子性。

嗯，这一块内容确实有点多。但你在实战中只要记住一句话：先定key，再选策略，最后看要不要事务。大部分场景，幂等性加合理的key设计就够用了。