Kafka- 探险 --- 生产者源码分析 --- 核心组件，Java 数据库题目大全

消息自然是通过应用程序构造出来并提供给生产者，生产者首先要知道需要将消息发送到哪个 Broker 的哪个 Topic，以及 Topic 的具体 Partition 。那么必然需要配置客户端的 Broker 集群地址 ，需要发送的 Topic 名称 ，以及 消息的分区策略 ，是指定到具体的分区还是通过某个 key hash 到不同的分区。

知道了消息要通往哪，还需要知道发送的是什么格式的消息，是字符串还是数字或是被序列化的二进制对象。 消息序列化 将需要消息序列化成字节数组才方便在网络上传输，所以要配置生产者的消息序列化策略，最好是可以通过传递枚举或者类名的方式自动构造序列化器，便于后续序列化过程的扩展。

消息队列常常用于多个系统之间的异步调用，那么这种调用关系就没有强实时依赖。由于发消息到 Kafka 会产生 网络 I/O ，相对来说比较耗时，那么消息发送这一动作除了同步调用， 是否也可以设置为异步，提高生产者的吞吐呢? 。并且大量消息发送场景, 我们可以设置一个窗口，窗口可以是时间维度也可以是消息数量维度，将消息积攒起来批次发送，减少网络 I/O 次数，提高吞吐量。

最后呢为了保证消息可以最大程度的成功发送到 Broker ，我们还需要一些 失败重试机制 ，例如失败后放到重试队列中，隔一段时间尝试再次发送。

此外，这里整理了一些 Kakfa笔记

理清思路

通过上面的分析，我们会有一个大致的认识，应该会有哪些方法，以及底层的大致的设计会分为哪几个部分。但是不够清楚，不够明晰。

首先总结一下实现客户端的几个要点在于：

1. 配置 Broker 基础信息：集群地址、Topic、Partition

2. 消息序列化，通过可扩展的序列化器实现

3. 消息异步写入缓冲区，网络 I/O 线程实现消息发送

4. 消息发送的失败重试机制

话不多说，用一张图画出各个核心模块以及他们之间的交互顺序：

![image](https://upload-images.jianshu.io/upload_ima

《Android学习笔记总结+最新移动架构视频+大厂安卓面试真题+项目实战源码讲义》
浏览器打开：qq.cn.hn/FTe 免费领取

ges/24613101-e0a3ca2f5c96097a.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

用户设定 Kafka 集群信息，生产者从 Kafka Broker 上拉取 可用 Kafka 节点、Topic 以及 Partition 对应关系。缓存到生产者成员变量中，如果 Broker 集群有扩容，或者有机器下线需要重新获取这些服务信息。

客户端根据用户设置的序列化器，对消息进行序列化，之后异步的将消息写入到客户端缓冲区。缓冲区内的消息到达一定的数量或者到达一个时间窗口后，网络 I/O 线程将消息从缓冲区取走，发送到 Broker 。

以上就是我对于一个 Kafka 生产者实现的思考，接下来看看官方的代码设计与我们的思路有何差别，他又是为什么这么设计。

官方设计

其实经过上面的思考和整理，我们的设计已经非常接近 Kafka 的官方设计了，官方的模块拆分的更加细致，功能更加独立。

核心组件

首先看一眼 KafkaProducer 类中有哪些成员变量，这些变量就是 Producer 的核心组件。

其中核心字段的解释如下：

clinetId ：标识发送者 Id

metric ：统计指标

partitioner ：分区器作用是决定消息发到哪个分区。有 key 则按照 key 的 hash ，否则使用 roundrobin

key/value Serializer ：消息 key/value 序列化器

interceptors ：发送之前/后对消息的统一处理

maxRequestSize ：可以发送的最大消息，默认值是 1M，即影响一个消息 Record 的大小，此值在服务端也是有限制的。

maxBlockTimeMs ：buffer 满了或者等待 metadata 信息的，超时的补偿机制

accumulator ：累积缓冲器

networkClient ：包装的网络层

sender ：网络 I/O 线程

发送流程

发送一条消息的时候，数据又是怎样在这些组件之间进行流转的呢？