生产者

生产者是 Kafka 消息的入口，理解生产者的工作原理是保证消息可靠投递的基础。

生产者架构

发送流程

┌─────────────────────────────────────────────────────────────┐
│                      Kafka Producer                          │
│                                                              │
│  ┌──────────┐    ┌────────────┐    ┌─────────────────────┐  │
│  │ 业务线程  │───▶│ Serializer │───▶│    Partitioner     │  │
│  │ send()   │    │  序列化器   │    │     分区器          │  │
│  └──────────┘    └────────────┘    └──────────┬──────────┘  │
│                                               │              │
│                                               ▼              │
│                              ┌────────────────────────────┐  │
│                              │      RecordAccumulator     │  │
│                              │   ┌─────┐ ┌─────┐ ┌─────┐ │  │
│                              │   │Batch│ │Batch│ │Batch│ │  │
│                              │   │ P0  │ │ P1  │ │ P2  │ │  │
│                              │   └─────┘ └─────┘ └─────┘ │  │
│                              └───────────────┬────────────┘  │
│                                              │               │
│  ┌────────────┐                              │               │
│  │  Sender    │◀─────────────────────────────┘               │
│  │  Thread    │                                              │
│  └─────┬──────┘                                              │
│        │                                                     │
└────────┼─────────────────────────────────────────────────────┘
         │
         ▼
    Kafka Broker

核心组件

组件	作用
Serializer	将消息 Key/Value 序列化为字节数组
Partitioner	决定消息发送到哪个 Partition
RecordAccumulator	消息累加器，批量发送
Sender	发送线程，负责网络 I/O
NetworkClient	网络客户端，管理连接

发送方式

三种发送方式

// 1. 发送并忘记（Fire and Forget）
producer.send(record);

// 2. 同步发送
RecordMetadata metadata = producer.send(record).get();
System.out.println("Partition: " + metadata.partition());
System.out.println("Offset: " + metadata.offset());

// 3. 异步发送（推荐）
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        exception.printStackTrace();
    } else {
        System.out.println("发送成功: " + metadata.offset());
    }
});

发送方式对比

方式	特点	适用场景
发送并忘记	不关心结果，可能丢消息	日志等可容忍丢失场景
同步发送	阻塞等待结果，性能差	对可靠性要求极高
异步发送	非阻塞，高性能	大多数场景

分区策略

默认分区策略

// Kafka 默认分区逻辑
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();

    if (keyBytes == null) {
        // 无 Key：轮询或粘性分区
        return stickyPartitionCache.partition(topic, cluster);
    } else {
        // 有 Key：hash 取模
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }
}

分区策略详解

1. 指定 Partition：直接发送到指定分区
   record = new ProducerRecord<>(topic, partition, key, value);

2. 有 Key：hash(key) % partitionCount
   - 相同 Key 总是发送到同一分区
   - 保证同一 Key 消息的顺序性

3. 无 Key：
   - 旧版本：轮询（Round Robin）
   - 新版本：粘性分区（Sticky Partitioner）
     - 批量发送到同一分区，提高批量效率

自定义分区器

public class OrderPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 按订单 ID 的 hash 分区
        String orderId = (String) key;
        int numPartitions = cluster.partitionCountForTopic(topic);
        return Math.abs(orderId.hashCode()) % numPartitions;
    }

    @Override
    public void close() {}

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

// 使用自定义分区器
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,
          OrderPartitioner.class.getName());

批量发送与压缩

批量发送

RecordAccumulator 工作原理：

┌────────────────────────────────────────────┐
│              RecordAccumulator             │
│                                            │
│  Partition 0    Partition 1    Partition 2 │
│  ┌────────┐     ┌────────┐     ┌────────┐ │
│  │ Batch  │     │ Batch  │     │ Batch  │ │
│  │ msg1   │     │ msg3   │     │ msg5   │ │
│  │ msg2   │     │ msg4   │     │        │ │
│  │ ...    │     │ ...    │     │        │ │
│  └────────┘     └────────┘     └────────┘ │
│       │              │              │      │
│       └──────────────┼──────────────┘      │
│                      │                     │
│                      ▼                     │
│              batch.size 或                 │
│              linger.ms 触发发送            │
│                                            │
└────────────────────────────────────────────┘

批量参数配置

# 批量大小（字节），达到此大小立即发送
batch.size=16384

# 等待时间（毫秒），即使未达到 batch.size 也发送
linger.ms=5

# 发送缓冲区总大小
buffer.memory=33554432

压缩配置

# 压缩算法：none, gzip, snappy, lz4, zstd
compression.type=lz4

压缩算法对比：

算法	压缩率	CPU 消耗	适用场景
none	无	无	CPU 敏感
gzip	最高	高	网络带宽受限
snappy	中等	低	通用
lz4	中等	最低	推荐
zstd	高	中等	Kafka 2.1+

关键配置参数

可靠性相关

# ACK 确认机制
# 0：不等待确认（可能丢消息）
# 1：等待 Leader 确认
# all/-1：等待所有 ISR 确认（最可靠）
acks=all

# 重试次数
retries=3

# 重试间隔
retry.backoff.ms=100

# 幂等性（防止重复）
enable.idempotence=true

性能相关

# 批量大小
batch.size=16384

# 等待时间
linger.ms=5

# 缓冲区大小
buffer.memory=33554432

# 最大请求大小
max.request.size=1048576

# 单个连接最大未确认请求数
# 设为 1 可保证顺序，但降低吞吐
max.in.flight.requests.per.connection=5

幂等性与事务

幂等性 Producer

// 开启幂等性
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

// 幂等性要求
// acks=all
// retries > 0
// max.in.flight.requests.per.connection <= 5

幂等性原理：

Producer ID (PID) + Sequence Number

┌──────────┐                    ┌──────────┐
│ Producer │                    │  Broker  │
│ PID=100  │                    │          │
└────┬─────┘                    └────┬─────┘
     │                               │
     │  msg1 (PID=100, Seq=0)       │
     │──────────────────────────────▶│ ✓ 接收
     │                               │
     │  msg2 (PID=100, Seq=1)       │
     │──────────────────────────────▶│ ✓ 接收
     │                               │
     │  重试 msg2 (PID=100, Seq=1)  │
     │──────────────────────────────▶│ ✗ 去重
     │                               │

事务 Producer

// 事务配置
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

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

// 初始化事务
producer.initTransactions();

try {
    // 开始事务
    producer.beginTransaction();

    // 发送消息
    producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic2", "key2", "value2"));

    // 提交事务
    producer.commitTransaction();
} catch (Exception e) {
    // 回滚事务
    producer.abortTransaction();
}

事务使用场景：

1. 跨 Topic 原子写入
2. Consume-Transform-Produce 模式
3. Exactly-Once 语义

完整代码示例

public class KafkaProducerExample {

    public static void main(String[] args) {
        Properties props = new Properties();

        // 基础配置
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                  StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                  StringSerializer.class.getName());

        // 可靠性配置
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        // 性能配置
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {

            for (int i = 0; i < 100; i++) {
                String key = "order-" + i;
                String value = "{\"orderId\": " + i + ", \"amount\": 100}";

                ProducerRecord<String, String> record =
                    new ProducerRecord<>("orders", key, value);

                // 异步发送
                producer.send(record, (metadata, exception) -> {
                    if (exception != null) {
                        System.err.println("发送失败: " + exception.getMessage());
                    } else {
                        System.out.printf("发送成功: topic=%s, partition=%d, offset=%d%n",
                                metadata.topic(), metadata.partition(), metadata.offset());
                    }
                });
            }

            // 确保所有消息发送完成
            producer.flush();
        }
    }
}

面试高频问题

1. 如何保证消息不丢失？

Producer 端：
1. acks=all：等待所有 ISR 副本确认
2. retries > 0：失败重试
3. 回调处理发送失败
4. 使用同步发送或确认异步结果

配置示例：
acks=all
retries=3
enable.idempotence=true

2. 如何保证消息有序？

1. 单分区有序：相同 Key 发送到同一分区
2. max.in.flight.requests.per.connection=1
   - 限制未确认请求数为 1
   - 避免重试导致乱序
3. 开启幂等性（enable.idempotence=true）
   - Kafka 2.0+ 支持最多 5 个未确认请求时保持顺序

3. batch.size 和 linger.ms 的关系？

发送触发条件（满足任一即发送）：
1. batch.size：批次达到指定大小
2. linger.ms：等待时间超过指定值

权衡：
- 大 batch.size + 长 linger.ms：高吞吐，高延迟
- 小 batch.size + 短 linger.ms：低延迟，低吞吐

4. 幂等性如何实现的？

原理：Producer ID + Sequence Number

1. Producer 初始化时获取 PID
2. 每条消息携带 <PID, Partition, Sequence>
3. Broker 记录每个 <PID, Partition> 的最大 Sequence
4. 收到重复 Sequence 的消息时丢弃

局限性：
- 单 Producer 单 Partition
- Producer 重启后 PID 变化
- 跨 Partition 需要事务

总结

Producer 核心要点：
1. 发送流程：序列化 → 分区 → 批量累积 → 发送
2. 分区策略：指定分区 > Key hash > 轮询/粘性
3. 批量发送：batch.size + linger.ms 权衡吞吐和延迟
4. 可靠性：acks=all + retries + 幂等性
5. 事务：跨 Topic 原子写入，Exactly-Once 语义