Henry's xyz

外观

Kafka Producer ​

发送流程 ​

ProducerRecord
  │
  ├── Serializer (Key → bytes, Value → bytes)
  ├── Partitioner (确定目标 Partition)
  │
  └── RecordAccumulator (内存缓冲区，按 Partition 分组)
        │
        └── Sender 线程
              │
              ├── 将批次（batch）发送到目标 Broker Leader
              ├── 等待 ACK 确认
              └── 失败 → 重试（retries）

基本用法 ​

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

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

ProducerRecord<String, String> record = new ProducerRecord<>("orders", "key-1", "value-1");
producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.printf("发送成功 → topic:%s, partition:%d, offset:%d%n",
            metadata.topic(), metadata.partition(), metadata.offset());
    }
});
shell
# 命令行生产
kafka-console-producer.sh --topic orders --bootstrap-server localhost:9092

分区策略 ​

策略配置行为
指定 Key 哈希默认partition = hash(key) % partition_count，相同 Key 进入同一 Partition
轮询（Round-Robin）key=null 时，v2.4+ StickyPartitioner将一批消息发往同一 Partition，然后切换，兼顾批量和均匀
自定义分区器partitioner.class=...实现 Partitioner 接口
java
// 自定义分区器
props.put("partitioner.class", "com.example.CustomPartitioner");

// 指定 Partition（绕过分区器）
new ProducerRecord<>("orders", 0, "key", "value");

// Sticky 分区（v2.4+，默认）
props.put("partitioner.class", "org.apache.kafka.clients.producer.StickyPartitioner");

投递语义 ​

语义ACKretries幂等说明
At-most-once00否不管是否成功，可能丢消息
At-least-once1 或 all> 0否默认，可能重复
Exactly-onceall> 0enable.idempotence=true5.0+ 默认，不丢不重

ACK 配置 ​

java
// acks=0 — 不等待确认，最高吞吐，会丢消息
props.put("acks", "0");

// acks=1 — Leader 写入即返回，可能丢消息（Leader 故障、Follower 未同步）
props.put("acks", "1");

// acks=all (或 -1) — 所有 ISR 确认后才返回，最安全
props.put("acks", "all");
acks吞吐延迟持久性适用
0最高最低无保证指标上报、日志（可容忍丢失）
1中中Leader 写入默认，一般场景
all / -1最低最高全 ISR 确认金融、支付、订单

幂等生产者 ​

java
// 开启幂等（Kafka 5.0 默认启用）
props.put("enable.idempotence", "true");
// 幂等要求: acks=all, retries > 0, max.in.flight.requests.per.connection ≤ 5
// 原理: Producer 分配 PID + 每个 Partition 维护 sequence number，Broker 去重

事务 ​

java
// 跨 Partition / Topic 原子写入
props.put("transactional.id", "txn-order-service-1");
producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", orderData));
    producer.send(new ProducerRecord<>("inventory", inventoryData));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

核心配置 ​

参数默认值说明
bootstrap.servers—Broker 地址，建议写多个
acksall（3.0+）确认级别
compression.typenone压缩：gzip / snappy / lz4 / zstd
batch.size16384批次字节数，达到后立即发送
linger.ms0等待时间，攒够批次再发（提升吞吐）
buffer.memory32MB缓冲区总大小
retries2147483647重试次数
max.in.flight.requests.per.connection5未确认的最大请求数
enable.idempotencetrue（3.0+）幂等生产者
java
// 高吞吐配置
props.put("compression.type", "lz4");     // 压缩
props.put("batch.size", 131072);          // 128KB 批次
props.put("linger.ms", 10);               // 攒 10ms
props.put("buffer.memory", 67108864);     // 64MB 缓冲区

// 低延迟配置
props.put("linger.ms", 0);
props.put("batch.size", 16384);

参考 ​