4. Kafka生产者实战：发送模式、分区策略、序列化器、拦截器、幂等性与事务

Kafka 生产者是数据流的源头，其配置与行为直接决定了消息投递的可靠性、吞吐量与顺序性。本章围绕生产者的六大核心机制展开实战讲解，帮助你在风控场景中精准控制每一条消息的流向与状态。

4.1 发送模式：三种投递语义

Kafka 生产者提供三种发送模式，分别对应不同的可靠性等级与吞吐量权衡：

// 异步回调示例
producer.send(new ProducerRecord<>("risk-events", key, value),
    (metadata, exception) -> {
        if (exception != null) {
            // 记录失败事件到本地缓存或死信队列
            log.error("发送失败: {}", exception.getMessage());
        } else {
            log.debug("发送成功: partition={}, offset={}",
                metadata.partition(), metadata.offset());
        }
    });

4.2 分区策略：数据路由的核心

分区决定了消息被写入哪个 Broker 的哪个物理文件。Kafka 默认提供三种分区策略，同时支持自定义实现。

• 默认分区器（DefaultPartitioner）：当 key 不为 null 时，对 key 的 hash 值取模；key 为 null 时，使用粘性分区（Sticky Partitioner）轮询。
• 轮询策略（RoundRobin）：消息均匀分布到所有分区，适合无 key 的日志类数据。
• 自定义分区器：实现 Partitioner 接口，按业务规则路由，例如按用户 ID 或风控标签路由到指定分区。

// 自定义分区器：按用户ID哈希路由
public class UserIdPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        // 使用用户ID的哈希值取模，确保同一用户进入同一分区
        return Math.abs(key.hashCode()) % numPartitions;
    }
}

4.3 序列化器：消息的编码与解码

生产者将 Java 对象转换为字节流发送到 Kafka，消费者再反序列化为对象。Kafka 内置了常用序列化器，也支持 Avro、Protobuf、JSON 等第三方序列化框架。

4.4 拦截器：消息发送的切面增强

生产者拦截器允许在消息发送前和发送完成后插入自定义逻辑，常用于埋点、监控、数据清洗或安全校验。

// 自定义拦截器：统计发送耗时并记录指标
public class MetricsInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 在消息头部注入发送时间戳
        record.headers().add("send-time", 
            String.valueOf(System.currentTimeMillis()).getBytes());
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            // 记录发送成功指标
            MetricsCollector.recordSuccess(metadata.topic());
        } else {
            MetricsCollector.recordFailure(metadata.topic(), exception);
        }
    }
}

4.5 幂等性：精确一次语义的基础

幂等性保证生产者发送的每条消息在 Broker 端仅被持久化一次，即使生产者重试也不会产生重复数据。其核心机制是 Producer ID (PID) + 序列号 (Sequence Number)。

• 开启方式：设置 enable.idempotence=true（默认 false）。
• 前提条件：acks 必须设置为 all，retries 必须大于 0。
• 限制：仅保证单分区内的精确一次，跨分区事务需要事务 API 支持。

// 开启幂等性配置
Properties props = new Properties();
props.put("enable.idempotence", true);
props.put("acks", "all");
props.put("retries", 3);
props.put("max.in.flight.requests.per.connection", 5);

4.6 事务：跨分区与跨会话的原子性

事务 API 允许生产者将多条消息（可跨分区、跨 Topic）作为一个原子单元提交，要么全部成功，要么全部回滚。事务在风控场景中常用于“先写事件、再写状态”的双写一致性需求。

// 事务 API 使用示例
producer.initTransactions();
try {
    producer.beginTransaction();
    // 发送多条消息到不同分区
    producer.send(new ProducerRecord<>("risk-events", "user1", event1));
    producer.send(new ProducerRecord<>("risk-events", "user2", event2));
    producer.send(new ProducerRecord<>("risk-status", "user1", status1));
    // 提交事务
    producer.commitTransaction();
} catch (Exception e) {
    // 回滚事务
    producer.abortTransaction();
    log.error("事务回滚: {}", e.getMessage());
}

4.7 实战对比：幂等性与事务的适用边界