【kafka实战】04 Kafka生产者发送消息过程源码分析

Kafka生产者发送消息过程源码分析

1. 概述

Kafka生产者（Producer）是Kafka系统中负责将消息发送到Kafka集群的客户端组件。生产者发送消息的过程涉及多个步骤，包罗消息的序列化、分区选择、消息累加、批次发送等。本文将深入分析Kafka生产者发送消息的源码，并团结相关原理图举行讲解。
Kafka 基本概念与术语

• Topic（主题）：Kafka 中的消息分类逻辑单位，类似于数据库中的表。生产者将消息发送到特定的主题，消费者则从相应主题订阅并接收消息。例如，在一个电商系统里，可以有 “订单主题” 用于通报订单相关信息，“用户行为主题” 记录用户浏览、购买等操作，不同范例的业务数据通过主题举行区分，方便管理与处理惩罚。
  
• Partition（分区）：主题进一步细分的物理存储单位，一个主题可以包含多个分区。分区的存在实现了数据的并行读写，提升了 Kafka 的吞吐量。每个分区在存储层面是一个有序的、不可变的消息序列，消息在分区内按照追加的方式写入，通过分区号来标识。好比一个拥有高并发写入需求的 “日志主题”，可以划分多个分区，让不同的日志数据分散到各个分区，避免单点写入瓶颈。
  
• Broker（署理）：Kafka 集群中的服务器实例，负责存储和转发消息。一个 Kafka 集群通常由多个 Broker 构成，它们协同工作，实现数据的高可用性与负载均衡。每个 Broker 都有本身的 ID，存储着主题的部分或全部分区数据，当生产者发送消息或消费者获取消息时，需要与 Broker 举行交互。
  
• Producer（生产者）：如前文所述，是消息的生产者，负责将外部系统的数据封装成消息，发送到 Kafka 集群的指定主题。它要处理惩罚消息序列化、缓冲、发送策略以及与集群的交互等诸多复杂任务，确保消息高效可靠传输，像及时数据收罗系统中的传感器数据收罗模块，就可以作为 Kafka 生产者将收罗到的数据推送给集群。
  
• Consumer（消费者）：与生产者相对，是从 Kafka 集群的主题中拉取消息并举行处理惩罚的客户端。消费者可以以不同的消费模式运行，如单个消费者独立消费、多个消费者构成消费组共同消费一个主题，消费组内的消费者通过分区分配策略，协同消费主题下的各个分区，实现数据的并行处理惩罚，常见于大数据及时分析场景，不同的分析任务作为消费者从相应主题获取数据举行运算。
  
• Consumer Group（消费组）：多个消费者构成的逻辑分组，主要用于实现消息的负载均衡与容错。同一消费组内的消费者不会重复消费同一个分区的消息，而是按照肯定策略分摊主题下各分区的消费任务，当组内某个消费者出现故障时，其他消费者能主动接管其负责的分区消费，包管数据处理惩罚的连续性，例如在一个大规模日志分析系统中，多个日志处理惩罚进程构成消费组，共同处理惩罚来自 “日志主题” 的海量数据。
  

2. Kafka生产者发送消息的核心流程

Kafka生产者发送消息的核心流程可以分为以下几个步骤：

• 消息创建与序列化：生产者创建消息对象，并将消息的键和值举行序列化。
  
• 分区选择：根据分区策略选择消息要发送到的分区。
  
• 消息累加：将消息添加到消息累加器（RecordAccumulator）中，等待批量发送。
  
• 批次发送：当满意肯定条件时，将消息批次发送到Kafka集群。
  
• 响应处理惩罚：处理惩罚Kafka集群返回的响应，确保消息发送乐成。
  

下面我们将团结源码详细分析每个步骤。
Kafka 生产者主要由以下几个紧张部分构成：

• RecordAccumulator：消息收集器，用于缓存待发送的消息。生产者会先将消息批量存入这里，而非一条条直接发送，以提升传输效率。
  
• Sender：真正负责将消息发送到 Kafka 集群的组件，它从 RecordAccumulator 中获取批量消息，并与集群建立连接，执行发送操作。
  
• Metadata：维护 Kafka 集群的元数据信息，例如集群中有哪些 broker，各个主题的分区分布等。生产者依据这些信息决定消息该发往那里。
  

3. 源码分析

3.1 消息创建与序列化

生产者发送消息的第一步是创建消息对象，并将消息的键和值举行序列化。Kafka消息的键和值可以是恣意范例的数据，但最终需要序列化为字节数组才能通过网络传输。

1. // org.apache.kafka.clients.producer.KafkaProducer#send
2. public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
3.     // 1. 序列化消息的键和值
4.     byte[] serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
5.     byte[] serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());
7.     // 2. 分区选择
8.     int partition = partition(record, serializedKey, serializedValue, cluster);
9.    
10.     // 3. 将消息添加到累加器
11.     RecordAccumulator.RecordAppendResult result = accumulator.append(record, serializedKey, serializedValue, headers, partition, maxTimeToBlock);
12.    
13.     // 4. 如果批次已满或创建了新批次，则唤醒发送线程
14.     if (result.batchIsFull || result.newBatchCreated) {
15.         this.sender.wakeup();
16.     }
17.    
18.     return result.future;
19. }

复制代码

在send方法中，起首通过序列化器将消息的键和值序列化为字节数组。Kafka提供了多种内置的序列化器，如StringSerializer、ByteArraySerializer等，用户也可以自定义序列化器。
3.2 分区选择

Kafka消息发送到哪个分区是由分区器（Partitioner）决定的。默认情况下，Kafka使用DefaultPartitioner，它根据消息的键举行哈希盘算，然后根据哈希值选择分区。如果消息没有键，则采用轮询的方式选择分区。

1. // org.apache.kafka.clients.producer.internals.DefaultPartitioner#partition
2. public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
3.     List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
4.     int numPartitions = partitions.size();
5.    
6.     if (keyBytes == null) {
7.         // 如果消息没有键，则采用轮询方式选择分区
8.         int nextValue = counter.getAndIncrement();
9.         List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
10.         if (availablePartitions.size() > 0) {
11.             return Utils.toPositive(nextValue) % availablePartitions.size();
12.         } else {
13.             return Utils.toPositive(nextValue) % numPartitions;
14.         }
15.     } else {
16.         // 如果消息有键，则根据键的哈希值选择分区
17.         return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
18.     }
19. }

复制代码

3.3 消息累加

消息累加器（RecordAccumulator）是Kafka生产者中的一个紧张组件，它负责将消息按分区举行缓存，并等待批量发送。每个分区对应一个消息批次（RecordBatch），当批次大小到达肯定阈值或等待时间超过肯定阈值时，批次会被发送到Kafka集群。

1. // org.apache.kafka.clients.producer.internals.RecordAccumulator#append
2. public RecordAppendResult append(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Header[] headers, int partition, long maxTimeToBlock) {
3.     // 获取或创建对应分区的Deque
4.     Deque<ProducerBatch> deque = getOrCreateDeque(partition);
5.    
6.     // 尝试将消息添加到批次中
7.     synchronized (deque) {
8.         RecordAppendResult result = tryAppend(deque, record, serializedKey, serializedValue, headers);
9.         if (result != null) {
10.             return result;
11.         }
12.     }
13.    
14.     // 如果批次已满或创建了新批次，则返回结果
15.     return appendNewBatch(deque, partition, record, serializedKey, serializedValue, headers, maxTimeToBlock);
16. }

复制代码

3.4 批次发送

当消息累加器中的批次满意发送条件时，发送线程（Sender）会将批次发送到Kafka集群。发送线程会从累加器中获取预备好的批次，并将其封装成ProducerRequest，然后通过网络发送到Kafka集群。

1. // org.apache.kafka.clients.producer.internals.Sender#run
2. public void run() {
3.     while (running) {
4.         // 从累加器中获取准备好的批次
5.         RecordAccumulator.ReadyCheckResult result = accumulator.ready(cluster, now);
6.         
7.         // 发送批次
8.         sendProduceRequests(result.readyNodes, now);
9.     }
10. }

复制代码

3.5 响应处理惩罚

Kafka集群在接收到消息后，会返回一个响应（ProducerResponse）。发送线程会处理惩罚这个响应，并根据响应结果更新消息的状态。如果消息发送乐成，则调用用户提供的回调函数（Callback）；如果发送失败，则根据配置的重试策略举行重试。

1. // org.apache.kafka.clients.producer.internals.Sender#handleProduceResponse
2. private void handleProduceResponse(ClientResponse response, Map<TopicPartition, ProducerBatch> batches, long now) {
3.     for (Map.Entry<TopicPartition, ProducerBatch> entry : batches.entrySet()) {
4.         TopicPartition tp = entry.getKey();
5.         ProducerBatch batch = entry.getValue();
6.         
7.         if (response.wasDisconnected()) {
8.             // 处理网络断开的情况
9.             handleDisconnection(batch, tp, now);
10.         } else if (response.hasResponse()) {
11.             // 处理成功响应
12.             handleSuccessfulResponse(batch, tp, response, now);
13.         } else {
14.             // 处理其他错误
15.             handleErrorResponse(batch, tp, response, now);
16.         }
17.     }
18. }

复制代码

4. 原理图

以下是Kafka生产者发送消息的核心流程示意图：

5. 总结

现将发送消息的详细流程总结如下

• 当应用程序调用生产者的 send 方法发送一条消息时，消息起首会被序列化。Kafka 支持多种序列化方式，如常见的 StringSerializer、AvroSerializer 等，确保消息能以符合的二进制格式传输。序列化后的消息被封装成 ProducerRecord 对象，该对象包含了目标主题、分区信息（若有指定）、键值对等关键数据。
  
• 接着，ProducerRecord 进入 RecordAccumulator。这里采用了一种类似缓冲池的机制，消息按照分区分类存放，每个分区都有本身独立的缓冲区。RecordAccumulator 会连续监测各个分区缓冲区的消息数目，一旦到达设定的批量大小（batch.size 参数配置），或者距离上次发送时间超过 linger.ms 设定的时长，就标志该分区的消息为可发送状态。
  
• Sender 线程不绝在背景运行，它周期性地轮询 RecordAccumulator，查找那些已标志为可发送的分区消息。当发现可发送的消息批次后，Sender 会从 Metadata 组件获取对应的 broker 地址信息，建立与目标 broker 的连接。这里涉及到 TCP 连接的建立、维护以及连接池的管理等复杂逻辑，以确保连接的高效复用与可靠性。
  
• 与 broker 乐成建立连接后，Sender 使用 Kafka 的协议格式，将消息批次组装成请求发送过去。在这个过程中，需要处理惩罚诸如请求超时、重试机制等非常情况。如果发送失败，根据生产者配置的重试次数（retries 参数），会主动举行重试，直到到达重试上限或者乐成为止。
  
• 一旦 broker 乐成接收并处理惩罚了消息批次，它会返回一个响应给生产者。Sender 负责解析这个响应，确认消息是否被准确写入 Kafka 日志文件。若出现错误，例如写入副本不足导致的写入失败，生产者大概会根据配置的策略举行调解，如增长重试次数、调解消息发送策略等。
  

Kafka生产者发送消息的过程涉及多个步骤，包罗消息的序列化、分区选择、消息累加、批次发送和响应处理惩罚。通过源码分析，我们可以更深入地明白Kafka生产者的工作原理。盼望本文能够资助你更好地明白Kafka生产者的内部机制。
6. 参考

• Kafka官方文档
  
• Kafka源码
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。