原理剖析| 一文搞懂 Kafka Producer（上）

前言

今天给大家带来的是 Kafka Producer 的全方位解析（基于 Apache Kafka 3.72）。考虑到篇幅限制，本文分为上下两篇，上篇将介绍 Kafka Producer 的使用方法与实现原理，下篇将介绍 Kafka Producer 的实现细节与常见问题。

使用方法

在介绍 Kafka Producer 的具体实现前，首先看一下如何使用。用 Kafka Producer 向指定 topic 发送一条消息的示例代码如下：

代码语言：shell复制

// 配置并创建一个 Producer
Properties kafkaProps = new Properties();
kafkaProps.put("bootstrap.servers", "localhost:9092");
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<>(kafkaProps);

// 向指定 topic 发送一条消息
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "my-key", "my-value");
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        // 发送失败
        exception.printStackTrace();
    } else {
        // 发送成功
        System.out.println("Record sent to partition "   metadata.partition()   " with offset "   metadata.offset());
    }
});

// 关闭 Producer，释放资源
producer.close();

接下来详细介绍一下 Kafka Producer 的主要接口。

代码语言：shell复制

public class ProducerRecord<K, V> {
    private final String topic;
    private final Integer partition;
    private final Headers headers;
    private final K key;
    private final V value;
    private final Long timestamp;
}

public interface Callback {
    void onCompletion(RecordMetadata metadata, Exception exception);
}

public interface Producer<K, V> {
    // ...
    Future<RecordMetadata> send(ProducerRecord<K, V> record);
    Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback);
    void flush();
    void close();
    // ...
}

注：在接口 Producer中，还有一些事务相关的接口，例如beginTransaction、commitTransaction等，其在我们另外一篇内容原理剖析| Kafka Exactly Once 语义实现原理：幂等性与事务消息中已经详细介绍过，此处不再赘述。

2.1 ProducerRecord

Producer 发送出的一条消息，包含以下属性

topic：必选。用于指定该 record 发送到的 topic
partition：可选。用于指定该 record 发送到的 partition 的序列号（从零开始编号，zero-indexed）。当未设置时，则使用用户指定的 Partitioner 或内置的 BuiltInPartitioner 选择分区（详见下文）
headers：可选。用户自定义的额外键值对信息
key：可选。消息的键值
value：可选。消息的内容
timestamp：可选。发送消息的时间戳。其生成逻辑为
如果 topic 的 message.timestamp.type 配置为 "CreateTime"
- 如果用户指定了 timestamp，则使用用户指定的值
- 若否，则使用创建这条消息的时间（约等于调用 send 方法的时间）
如果 topic 的 message.timestamp.type 配置为 "LogAppendTime"，则无论用户是否指定了 timestamp，都使用消息在 broker 上写入时的时间

2.2 Callback

用于发送消息 ack 后的回调。可能发生的 Exception 有：

不可重试
InvalidTopicException：topic 的名称不合法，例如过长、为空、使用非法字符等
OffsetMetadataTooLarge：调用 Producer#sendOffsetsToTransaction 时，使用的 Metadata 字符串过长（由 offset.metadata.max.bytes 控制，默认 4 KiB）
RecordBatchTooLargeException：发送的 batch 的大小
- 超过了允许的最大大小（broker 配置 message.max.bytes 或 topic 配置 max.message.bytes，默认 1MiB 12 B）
- 超过了 segment 的大小（broker 配置 log.segment.bytes 或 topic 配置 segment.bytes，默认 1 GiB）undefined注：该错误仅可能会发生在老版本的 Client 中
RecordTooLargeException：单条消息的大小超过了 producer 单个请求的最大大小（producer 配置 max.request.size，默认 1MiB）
TopicAuthorizationException、ClusterAuthorizationException：鉴权失败
UnknownProducerIdException：事务请求中，PID 已过期或 PID 关联的 record 均已过期
InvalidProducerEpochException：事务请求中，epoch 非法
UnknownServerException：未知错误
可重试
CorruptRecordException：CRC 校验失败，通常由网络错误导致
InvalidMetadataException：Client 侧的 metadata 过期
- UnknownTopicOrPartitionException：topic 或 partition 不存在，可能由 metadata 过期导致
- NotLeaderOrFollowerException：请求的 broker 不是 leader，可能正在选举 leader
- FencedLeaderEpochException：请求中的 leader epoch 过期，可能由 metadata 刷新慢导致
NotEnoughReplicasException、NotEnoughReplicasAfterAppendException：insync replica 数量不足（broker 配置 min.insync.replicas 或同名 topic 配置，默认 1）。注意，NotEnoughReplicasAfterAppendException 会在 record 写入完成后发现，producer 的重试会导致数据重复
TimeoutException：处理超时，有两种可能
同步调用耗时过长，例如 producer buffer 满、拉取 metadata 超时等
异步调用超时，例如 producer 被限流导致没有发送、broker 超时未响应等

2.3 Producer#send

异步地发送一条消息，如果需要，在本条消息 ack 后触发 Callback。

保证向同一个 partition 发送的 send 请求的 Callback 会按调用顺序依次触发。

2.4 Producer#flush

标记 producer 缓存中的所有消息立即可用于发送，并阻塞当前线程，直至在此之前的所有消息都被 ack。

注：仅会阻塞当前线程，其他线程仍可正常发送，但对调用 flush 方法后发送的其他消息的完成时机没有保证。

2.5 Producer#close

关闭 producer，并阻塞等待至所有消息发送完成。

注：

在 Callback 中调用 close 会立刻关闭 producer
仍处于同步调用阶段（拉取 metadata、等待分配内存）的 send 方法将会立即终止，并抛出 KafkaException

核心组件

接下来介绍 Kafka Producer 的具体实现，它由以下几个核心组件组成

ProducerMetadata & Metadataundefined负责 Producer 侧所需元数据的缓存与刷新，其中包含 Kafka Cluster 的所有元数据，例如 broker 地址、topic 中的 partition 的分布状态、leader 与 follower 信息。
RecordAccumulatorundefined负责维护 Producer 的缓冲区。它会将待发送的消息按照 partition 的维度、基于时间（linger.ms）和空间（batch.size）攒批为 RecordBatch，并等待发送。
Senderundefined维护一个守护线程 "kafka-producer-network-thread | {client.id}"，负责驱动发送 Produce 请求和处理 Produce 响应，同时负责超时处理、错误处理与重试。
TransactionManagerundefined负责实现幂等（idempotence）与事务（transaction）。包括分配序号（sequence number）、处理消息丢失与乱序、维护事务状态等。

发送流程

一条消息的发送流程如下图：

分为以下几步：

刷新元数据；
使用指定的 Serializer 序列化消息；
使用用户指定 Partitioner 或 BuiltInPartitioner 选择发送消息的目标 partition；
将消息插入到 RecordAccumulator 进行攒批；
Sender 异步地从 RecordAccumulator 中取出可发送的 batch（按照 node 分组），注册回调，并发送；
Sender 处理响应，并根据情况返回结果、返回异常或重试。

接下来介绍其中的各项细节

4.1 刷新元数据

ProducerMetadata负责 Producer 侧所需元数据的缓存与刷新，它会维护一个 topic 视图，其中包含 producer 所需的所有 topic。它会

在以下场景增加 topicundefined当发送消息时，指定的 topic 不在缓存的元数据中
在以下场景移除 topicundefined当发现某个 topic 的元数据已经持续 metadata.max.idle.ms 未被使用时
在以下场景刷新元数据undefined当发送消息时，指定的 partition 不在缓存的元数据中（这会发生在 topic 的 partition 数量增加时）undefined当发送消息时，指定的 partition 的 leader 未知undefined当发送消息后，收到了 InvalidMetadataException 响应undefined当持续 metadata.max.age.ms 未刷新元数据时

相关配置有

partitioner.class

分区选择器的类名，可以由用户根据需求自行实现。提供了一些默认实现

DefaultPartitioner 与 UniformStickyPartitioner：会 "sticky" 地向各 partition 分配消息，即，在某个 partition 攒满一个 batch 后，切换至下一个 partition。但其实现上存在问题，会导致向更慢的 broker 发送更多消息，现已标记为废弃。
RoundRobinPartitioner：将会忽略 record key，循环（round robin）地向每个 partition 分配消息。注意，它存在一个已知问题：在创建新的 batch 时，会导致不平均的分配。目前建议使用内置 partitioner 或者自行实现 partitioner。
partitioner.adaptive.partitioning.enable

是否根据 broker 的速度决定发送消息的数量，若不开启，则会随机地选择 partition。仅在未配置 partitioner.class 时生效。默认为 "true"。

partitioner.availability.timeout.ms

仅在 partitioner.adaptive.partitioning.enable 设置为 "true" 时生效。当“为指定 broker 攒出一批消息的时间点”和“向指定 broker 发送消息的时间点”相差超过此配置时，则不再向指定 broker 分配消息；设置为 0 意味着不开启此逻辑。仅在未配置 partitioner.class 时生效。默认为 0。

partitioner.ignore.keys

选择 partition 时是否忽略消息的 key，若为 "false"，则根据 key 的哈希值选择 partition，否则忽略 key 值。仅在未配置 partitioner.class 时生效。默认为 "false"。

4.3 消息攒批

在 RecordAccumulator 中，按照 partition 维度维护了所有待发送的 batch。有以下几个重要方法：

代码语言：shell复制

 public RecordAppendResult append(String topic,
                                 int partition,
                                 long timestamp,
                                 byte[] key,
                                 byte[] value,
                                 Header[] headers,
                                 AppendCallbacks callbacks,
                                 long maxTimeToBlock,
                                 boolean abortOnNewBatch,
                                 long nowMs,
                                 Cluster cluster) throws InterruptedException;

public ReadyCheckResult ready(Metadata metadata, long nowMs);

public Map<Integer, List<ProducerBatch>> drain(Metadata metadata, Set<Node> nodes, int maxSize, long now);

append：将消息插入到缓冲区，注册一个 future 并返回，该 future 会在消息发送完成（成功或失败）时完成。
ready：筛选出所有拥有可发送消息的 node 列表。有以下几种情况：undefined已经攒批出 batch.size 大小的消息undefined已经持续攒批超过了 linger.ms 时间undefined分配给 producer 的内存已耗尽，即，缓冲区的消息大小总和超过了 buffer.memoryundefined需要重试的 batch 已经等待至少 retry.backoff.ms 时间undefined用户调用了 Producer#flush 以强制发送消息undefined正在关闭 producer
drain：对于每个 node，遍历其上的每个 partition，取出每个 partition 上最早的 batch（如果有），直至攒够 max.request.size 大小的消息，或遍历完所有 partition