优享资讯 | 原理剖析| 一文搞懂 Kafka Producer（上）

掘金后端 ( ) • 2024-05-17 10:12

前言

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

使用方法

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

// 配置并创建一个 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 的主要接口。

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

发送流程

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

分为以下几步：

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

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

4.1 刷新元数据

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

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