Kafka - 图解生产者消息发送流程

---

发送原理

---

Kafka的Producer发送消息采用的是异步发送的方式。

在消息发送的过程中，涉及到了两个线程:main线程和Sender线程，以及一个线程共享变量:RecordAccumulator。

• ①main线程中创建了一个双端队列RecordAccumulator，将消息发送给RecordAccumulator。

• ②Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。

---

Kafka的Producer发送消息采用了异步发送的方式，这个过程确实涉及到多个线程以及共享变量。下面详细展开说明这个过程：

1. 主线程 (main thread)：

主线程是生产者应用的线程，它负责创建消息并将这些消息发送给Kafka Producer API。主要的操作包括：

• 创建消息：主线程创建消息，将它们封装成ProducerRecord对象。ProducerRecord通常包括消息的主题（topic）、分区（partition）、键（key）和值（value）等信息。

• 发送消息到RecordAccumulator：主线程将创建的消息发送到一个双端队列（deque）叫做RecordAccumulator。这个队列用于缓冲消息，允许Producer线程将消息异步发送到Kafka集群，而不需要等待每条消息都被立刻发送。

2. Sender 线程：

Sender线程是Kafka Producer内部的一个后台线程，它负责从RecordAccumulator中拉取消息并发送到Kafka broker。Sender线程的主要工作如下：

• 从RecordAccumulator拉取消息：Sender线程定期轮询（poll）RecordAccumulator，检查是否有新消息需要发送。这个轮询是异步的，因此主线程不需要等待消息被发送。

• 构建请求：当Sender线程发现有消息需要发送，它会构建一个或多个ProducerRequest，每个请求包含多个消息，以便进行有效的批量发送。

• 发送消息到Kafka broker：Sender线程将构建的请求发送到Kafka broker，等待来自broker的响应。一旦消息被成功接收并记录在Kafka broker中，Sender线程会通知RecordAccumulator，以便它可以更新消息的状态。

3. RecordAccumulator：

RecordAccumulator是Producer内部的一个共享变量，用于暂存即将发送到Kafka broker的消息。主要功能包括：

• 暂存消息：主线程将消息发送到RecordAccumulator中，使其在等待Sender线程处理。

• 管理消息的状态：RecordAccumulator跟踪每条消息的发送状态，以确保消息被成功发送到Kafka broker。一旦消息被成功写入到Kafka broker的日志中，RecordAccumulator会将消息的状态标记为已发送。

• 负责消息批量化：RecordAccumulator也有助于消息的批量发送，以减少网络开销和提高性能。

发送原理小结

总结一下，Kafka的Producer采用异步发送消息的方式，

• 主线程负责创建和发送消息到RecordAccumulator，

• 而Sender线程负责从RecordAccumulator中拉取消息并将其发送到Kafka broker。

• RecordAccumulator充当缓冲区，用于管理消息的状态以及批量发送，以提高性能和降低延迟。

这个架构充分利用了多线程和异步操作，使得Producer能够高效地发送消息到Kafka集群。

---

重要参数