02. Kafka 生产者 API 与发送语义

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 可靠性
props.put("acks", "all");              // 0/1/all
props.put("retries", Integer.MAX_VALUE);  // 无限重试
props.put("enable.idempotence", true);  // 幂等性（默认false）

// 性能
props.put("linger.ms", 5);             // 批量等待 5ms
props.put("batch.size", 16384);        // 批量大小 16KB
props.put("compression.type", "lz4");  // 压缩算法

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

// 异步发送（推荐）
producer.send(new ProducerRecord<>("orders", orderId, json),
    (metadata, exception) -> {
        if (exception != null) exception.printStackTrace();
    });

// 同步发送
RecordMetadata metadata = producer.send(record).get();

// 带事务的发送（Exactly-Once）
producer.initTransactions();
producer.beginTransaction();
producer.send(record1);
producer.send(record2);
producer.commitTransaction();

// 默认: murmur2(key) % numPartitions
// 无 key: round-robin
// 自定义:
props.put("partitioner.class", CustomPartitioner.class.getName());

public class CustomPartitioner implements Partitioner {
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int numPartitions = cluster.partitionCountForTopic(topic);
        // 同一用户的消息到同一分区（保证顺序）
        return Math.abs(key.hashCode()) % numPartitions;
    }
}