4. Kafka生产者实战：生产者API、发送模式、分区策略与序列化器

在实时风控系统中，Kafka生产者是数据流的起点——它将风控事件（如登录、交易、授信请求）可靠地发送到Kafka集群。本章聚焦生产者的核心实战能力：API调用、发送模式选择、分区策略定制以及序列化器配置。

4.1 生产者核心API与初始化

Kafka生产者通过 KafkaProducer<K, V> 对象完成消息发送。初始化时需配置至少三个关键参数：bootstrap.servers、key.serializer、value.serializer。

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092,node3:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 风控场景推荐开启幂等性
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

4.2 发送模式：同步 vs 异步

Kafka提供三种发送模式，风控系统需根据延迟与可靠性要求权衡选择。

• 发后即忘（Fire-and-Forget）：不关心发送结果，吞吐最高但可能丢失消息。风控场景禁止使用。
• 同步发送（Sync）：调用 producer.send().get() 阻塞等待Broker确认。延迟高，但能立即捕获异常。
• 异步发送（Async）：传入 Callback 回调，非阻塞。风控系统推荐模式，兼顾吞吐与可靠性。

// 同步发送（阻塞）
try {
    RecordMetadata metadata = producer.send(record).get();
    System.out.println("同步发送成功，offset: " + metadata.offset());
} catch (Exception e) {
    System.err.println("同步发送失败: " + e.getMessage());
}

// 异步发送（风控推荐）
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        // 记录失败事件到本地日志或死信队列
        log.error("风控事件发送失败: {}", record.value(), exception);
        // 可触发告警或降级
    } else {
        log.info("事件发送成功, topic={}, partition={}, offset={}",
                 metadata.topic(), metadata.partition(), metadata.offset());
    }
});

4.3 分区策略：从默认到自定义

分区决定了消息被写入哪个Partition，直接影响消费并行度与顺序性。风控系统通常需要按用户ID或设备指纹进行分区，保证同一用户的事件顺序消费。

4.3.1 默认分区器（DefaultPartitioner）

• Key不为null：对Key的哈希值取模（Murmur2算法），相同Key进入同一分区。
• Key为null：使用粘性分区策略（Sticky Partitioner），批量填充一个分区后再切换，提升吞吐。

4.3.2 自定义分区器

当默认策略无法满足风控需求时（如需要按风控等级分流、按地域隔离），可实现 Partitioner 接口。

public class RiskLevelPartitioner implements Partitioner {
    private static final Map<String, Integer> LEVEL_MAPPING = Map.of(
        "HIGH", 0,   // 高风险事件写入分区0
        "MEDIUM", 1, // 中风险事件写入分区1
        "LOW", 2     // 低风险事件写入分区2
    );

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        String riskLevel = extractRiskLevel(value); // 从消息体中解析风控等级
        return LEVEL_MAPPING.getOrDefault(riskLevel, 2);
    }

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

4.4 序列化器：从内置到自定义

序列化器将Java对象转换为字节流。Kafka内置了String、Integer、ByteArray等序列化器，但风控事件通常是复杂的POJO（如包含用户ID、IP、设备指纹、金额等字段），需要自定义序列化器或使用通用序列化框架。

4.4.1 内置序列化器对比

4.4.2 自定义序列化器（Avro示例）

风控系统推荐使用 Avro 或 Protobuf 作为序列化方案，支持Schema注册与兼容性检查。以下为Avro序列化器实现片段：

public class RiskEventAvroSerializer implements Serializer<RiskEvent> {
    private final DatumWriter<RiskEvent> writer;
    private final EncoderFactory encoderFactory;

    public RiskEventAvroSerializer() {
        this.writer = new SpecificDatumWriter<>(RiskEvent.class);
        this.encoderFactory = EncoderFactory.get();
    }

    @Override
    public byte[] serialize(String topic, RiskEvent data) {
        if (data == null) return null;
        try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
            BinaryEncoder encoder = encoderFactory.binaryEncoder(out, null);
            writer.write(data, encoder);
            encoder.flush();
            return out.toByteArray();
        } catch (IOException e) {
            throw new SerializationException("Avro序列化风控事件失败", e);
        }
    }
}

4.5 生产者端到端流程总览

4.6 实战要点总结

• 初始化：必须配置 bootstrap.servers、序列化器；风控场景强制开启 enable.idempotence=true 和 acks=all。
• 发送模式：使用异步发送 + Callback回调，在回调中处理失败事件（写入死信队列或本地日志）。
• 分区策略：默认按Key哈希分区；风控系统建议自定义分区器，按用户ID或风控等级分流。
• 序列化器：优先选择Avro/Protobuf等带Schema的序列化框架，便于风控事件字段演进。
• 监控指标：关注 record-error-rate、record-queue-time-avg、buffer-available-bytes，及时发现生产者异常。