3. Kafka 生产者：原理、发送模式、分区策略与自定义 Partitioner

在风控网关数据链路中，Kafka 生产者是事件流的起点。风控决策依赖实时、有序、可靠的事件传输，生产者的配置直接决定数据能否被下游消费者正确消费。本节深入剖析生产者的核心原理、三种发送模式、分区策略以及自定义 Partitioner 的实现方法。

3.1 生产者核心原理

Kafka 生产者客户端负责将消息发送到指定的 Topic。其内部工作流程可概括为以下步骤：

1. 消息封装：将业务数据（如风控事件 JSON）封装为 ProducerRecord 对象，包含 Topic、Partition（可选）、Key、Value 和时间戳。
2. 序列化：通过 Serializer 将 Key 和 Value 转换为字节数组。默认使用 StringSerializer，风控场景常使用 Avro 或 Protobuf 序列化器以保障 schema 兼容性。
3. 分区器：根据 Key 和分区策略确定消息发往哪个 Partition。默认分区器在 Key 不为 null 时使用 murmur2 哈希取模，Key 为 null 时使用轮询（Round-Robin）。
4. 缓冲区：消息进入 RecordAccumulator（默认 32MB），按分区缓存为批次（batch）。
5. Sender 线程：后台线程从缓冲区拉取就绪批次，通过 Selector 网络连接发送到 Broker。
6. 响应处理：根据 acks 配置等待 Broker 确认，触发回调或返回 Future。

3.2 发送模式：同步、异步与 Fire-and-Forget

Kafka 生产者提供三种发送模式，风控场景需根据对可靠性和延迟的要求选择：

3.2.1 异步发送回调示例

// 异步发送带回调
producer.send(new ProducerRecord<>("risk-events", key, eventJson),
    new Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception != null) {
                // 记录失败事件到本地缓冲区或死信队列
                log.error("发送风控事件失败: {}", exception.getMessage());
                deadLetterQueue.offer(eventJson);
            } else {
                log.debug("事件发送成功, partition={}, offset={}",
                    metadata.partition(), metadata.offset());
            }
        }
    });

3.3 分区策略

分区策略决定消息在 Topic 内部分布到哪个 Partition。合理的分区策略对风控场景至关重要：

• 按用户 ID 分区：保证同一用户的所有事件进入同一分区，实现有序处理。
• 按事件类型分区：高优先级事件（如风控告警）集中到少数分区，便于消费者优先处理。
• 按设备指纹分区：同一设备的事件顺序消费，用于设备画像和关联分析。

3.4 序列化与自定义 Partitioner

3.4.1 序列化器选择

Kafka 内置多种序列化器，风控场景推荐使用 schema 演化友好的序列化方式：

3.4.2 自定义 Partitioner 实现

当默认分区策略无法满足业务需求时，可实现 Partitioner 接口。以下示例展示一个按风控事件优先级分区的 Partitioner：

public class RiskPriorityPartitioner implements Partitioner {
    private static final Map<String, Integer> PRIORITY_MAP = new HashMap<>();
    static {
        PRIORITY_MAP.put("HIGH", 0);   // 高优先级事件发往 partition 0
        PRIORITY_MAP.put("MEDIUM", 1); // 中优先级事件发往 partition 1
        PRIORITY_MAP.put("LOW", 2);    // 低优先级事件发往 partition 2
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 从 value 中解析优先级字段（假设 JSON 格式）
        String priority = extractPriority((String) value);
        Integer partition = PRIORITY_MAP.get(priority);
        if (partition != null) {
            return partition;
        }
        // 默认使用哈希
        return Utils.toPositive(Utils.murmur2(keyBytes)) % cluster.partitionCountForTopic(topic);
    }

    private String extractPriority(String jsonValue) {
        // 简单 JSON 解析，生产环境建议使用 Jackson
        if (jsonValue.contains("\"priority\":\"HIGH\"")) return "HIGH";
        if (jsonValue.contains("\"priority\":\"MEDIUM\"")) return "MEDIUM";
        return "LOW";
    }

    @Override
    public void close() {}

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

3.5 生产者配置最佳实践（风控场景）