多图详解 kafka 生产者消息发送过程

生产者客户端代码

public class SzzTestSend {    public static final String bootStrap = "xxxxxx:9090";    public static final String topic = "t_3_1";    public static void main(String[] args) {        Properties properties = new Properties();        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootStrap);        // 序列化协议  下面两种写法都可以        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");        //过滤器 可配置多个用逗号隔开        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"org.apache.kafka.clients.producer.SzzProducerInterceptorsTest");        //构造 KafkaProducer        KafkaProducer producer = new KafkaProducer(properties);        //  发送消息, 并设置 回调(回调函数也可以不要)        ProducerRecord<String,String> record = new ProducerRecord(topic,"Hello World!");        try {            producer.send(record,new SzzTestCallBack(record.topic(), record.key(), record.value()));        }catch (Exception e){            e.printStackTrace();        }    }    /**     * 发送成功回调类     */    public static class SzzTestCallBack implements Callback{        private static final Logger log = LoggerFactory.getLogger(SzzTestCallBack.class);        private String topic;        private String key;        private String value;        public SzzTestCallBack(String topic, String key, String value) {            this.topic = topic;            this.key = key;            this.value = value;        }        public void onCompletion(RecordMetadata metadata, Exception e) {            if (e != null) {                log.error("Error when sending message to topic {} with key: {}, value: {} with error:",                        topic, key,value, e);            }else {                log.info("send message to topic {} with key: {} value:{} success, partiton:{} offset:{}",                        topic, key,value,metadata.partition(),metadata.offset());            }        }    }}

1 构造 KafkaProducer

KafkaProducer 通过解析

producer.propeties

文件里面的属性来构造自己。例如 ：分区器、Key 和 Value 序列化器、拦截器、RecordAccumulator消息累加器 、元信息更新器、启动发送请求的后台线程

        //构造 KafkaProducer        KafkaProducer producer = new KafkaProducer(properties);

生产者元信息更新器

我们之前有讲过. 客户端都会保存集群的元信息,例如生产者的元信息是 ProducerMetadata. 消费组的是 ConsumerMetadata 。

相关的 Producer 配置有:

虽然 Producer 元信息会自动更新, 但是有可能在生产者发送消息的时候,发现某个 TopicPartition 不存在,这个时候可能就需要立刻发起一个元信息更新了。

集群资源变更监听器

org.apache.kafka.common.ClusterResourceListener

在构造 **KafkaConsumer **的时候, 还会构造一个 集群资源变更监听器 ClusterResourceListener

当用户希望收到有关集群元数据更改的通知时，可以实现回调接口。

需要在拦截器、指标采样器、序列化器和反序列化器 中访问集群元数据的用户可以实现此接口。

public interface ClusterResourceListener {    /**     * 用户可以实现以获取 ClusterResource 更新的回调方法。     * @param clusterResource cluster metadata     */    void onUpdate(ClusterResource clusterResource);}

下面描述了每种类型的方法调用顺序。

Clients

在每个元数据响应之后都会调用一次

onUpdate(ClusterResource)

当在

org.apache.kafka.clients.producer.ProducerInterceptor

实现的 ClusterResourceListener 的时候

调用顺序为: ProducerInterceptor.onSend() -> onUpdate(ClusterResource) -> ProducerInterceptor.onAcknowledgement()

当在

org.apache.kafka.clients.consumer.ConsumerInterceptor

实现的 ClusterResourceListener 的时候

调用顺序为：onUpdate() - > ConsumerInterceptor.onConsume()

当在

org.apache.kafka.common.serialization.Serializer

实现的 ClusterResourceListener 的时候

调用顺序为：onUpdate() - > Serializer.serialize(String, Object)

当在

org.apache.kafka.common.serialization.Deserializer

实现的 ClusterResourceListener 的时候

调用顺序为：onUpdate() - > .Deserializer.deserialize(String, byte[])

生产者拦截器

生产者拦截器在消息发送之前可以做一些准备工作, 比如 按照某个规则过滤某条消息, 又或者对 消息体做一些改造, 还可以用来在发送回调逻辑之前做一些定制化的需求,例如统计类的工作! 拦截器的执行时机在最前面,在消息序列化和分区计算之前

相关的 Producer 配置有:

生产者分区器

用来设置发送的消息具体要发送到哪个分区上

相关的 Producer 配置有:

Sender 线程启动

Sender 是专门负责将消息发送到 Broker 的 I/O 线程。

相关的 Producer 配置有:

2 发送请求

            producer.send(record,new SzzTestCallBack(record.topic(), record.key(), record.value()));            

生产者拦截器

发送消息的第一步就是执行拦截器

在这里插入图片描述

一般情况下我们可能不需要拦截器, 但是我们需要用拦截器的时候按照下面操作执行:

1. 在配置文件中配置属性interceptor.classes=拦截器1,拦截器2,拦截器3
2. 实现接口org.apache.kafka.clients.producer.ProducerInterceptor<K, V>

这个

interceptor.classes

中的属性可以配置多个拦截器, 用逗号隔开,并且执行顺序就是按照配置的顺序执行的。

拦截器的执行时机在最前面,在消息序列化和分区计算之前

ProducerInterceptor

org.apache.kafka.clients.producer.ProducerInterceptor<K, V>

接口方法讲解：

  public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);    public void onAcknowledgement(RecordMetadata metadata, Exception exception);    public void close();

onSend(ProducerRecord<K, V> record)方法 :

当客户端将记录发送到

KafkaProducer

时，在键和值被序列化之前调用。 该方法调用

ProducerInterceptor.onSend(ProducerRecord)

方法。 从第一个拦截器的

onSend()

返回的

ProducerRecord

传递给第二个拦截器 onSend()，在拦截器链中依此类推。 从最后一个拦截器返回的记录就是从这个方法返回的。 此方法不会抛出异常。 任何拦截器方法抛出的异常都会被捕获并忽略。 如果链中间的拦截器（通常会修改记录）抛出异常，则链中的下一个拦截器将使用前一个未抛出异常的拦截器返回的记录调用。

调用地方

①. 拦截器执行时机在键值序列化之前②. 拦截器抛出异常会被捕获,并打印日志,那么也意味着这个拦截器所做的修改不会生效③.拦截器中修改的消息体会被传递到下一个拦截器

onAcknowledgement(RecordMetadata metadata, Exception exception)方法:

当发送到服务器的记录已被确认时，或者当发送记录在发送到服务器之前失败时，将调用此方法。此方法通常在用户设置的 Callback 之前调用，此方法不会抛出异常。 任何拦截器方法抛出的异常都会被捕获并忽略。这个方法运行在 Producer 的 I/O 线程中,所以这个方法中的代码逻辑需要越简单越好。 否则，来自其他线程的消息发送可能会延迟。

参数：

metadata

– 已发送记录的元数据（即分区和偏移量）。 如果发生错误，元数据将只包含有效的主题和分区。 如果 ProducerRecord 中没有给出 partition 并且在分配 partition 之前发生错误，则 partition 将设置为 RecordMetadata.NO_PARTITION。 如果客户端将空记录传递给 KafkaProducer.send(ProducerRecord)则元数据可能为空。

exception

– 在处理此记录期间抛出的异常。 如果没有发生错误，则为空。

close()

主要用于在关闭拦截器时自行一些资源清理工作。

configure(Map<String, ?> configs)

ProducerInterceptor

接口中集成了一

Configurable

接口,接口有个方法

    void configure(Map<String, ?> configs);

也就是说在拦截器中,我们可以拿到所有的配置属性了; 这个方法在这几个方法中最早执行

生产者拦截器示例

将发送的消息加上后缀注意这里消息 value 的类型是

String

，如果是 byte 则需要处理一下

    @Override    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {        System.out.println("生产者拦截器 onSend()  run ."+record);        return new ProducerRecord<>(                record.topic(), record.partition(), record.key(), record.value().concat("_后缀"));    }    @Override    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {        System.out.println("生产者拦截器 onAcknowledgement run ."+metadata.toString() +" exception:"+exception);    }    @Override    public void close() {        System.out.println("生产者拦截器 close()  run .");    }    @Override    public void configure(Map<String, ?> configs) {        this.configs = configs;        System.out.println("生产者拦截器 configure  run ."+configs);    }

复制代码

更新元信息 waitOnMetadata

在发送消息之前,要先获取一下将要发送的 TopicPartition 的元信息。这个获取元信息的请求也是通过唤醒 **Sender **线程进行发送的。

1 .

ProducerMetadata

元信息

Map<String, Long> topics

中保存

Topic

的有效期时间,

metadata.max.idle.ms

控制,默认

300000

2.

ProducerMetadata

元信息

Set<String> newTopics

中保存所有

Topic

1. 获取 Topic 的

   元数据集群
   

   以及我们等待的时间（以毫秒为单位）, 这个获取元数据不是这里获取的,这里只是判断当前是否已经获取到了元数据,如果没有获取到,则一直等待,最大等待时间由

   max.block.ms
   

   控制,默认 60000(1 分钟),关于获取元数据在最上面已经分析过了, 是 Sender 线程获取并更新的。如果等待时间超过了

   max.block.ms
   

   ,很有可能网络异常,那么会抛出超时异常。4. 当你发送消息的时候指定了分区号, 但是这个分区号是不存在的, 这个时候就会一直发起

   Metadata
   

   请求(流程看最上面), 直到超时(

   max.block.ms
   

   )之后 抛出异常

org.apache.kafka.common.errors.TimeoutException: Topic t_3_1 not present in metadata after 60000 ms.

复制代码

相关的 Producer 配置有:

KafkaProducer producer = new KafkaProducer(properties);

在构建

KafkaProducer

对象的时候, 有构建

producer I/O thread

, 并且启动了,

Runnable

是

sender

最终调用

NetworkClient.poll(long timeout, long now)

里面

maybeUpdate()

方法这个方法会获取 前 Node 中负载最少的节点发起网络请求, 如果所有 Node 都是满负载则请求不会被发起。

如何判断哪个节点负载最少？

通过每个节点的

InFlightRequests(空中请求数量)

里面的最小数量判断,这个表示当前正在发起的请求,但是还没有收到回复的请求数量; 保存形式是一个 HashMap,

key

是 Node 的 Id,

value

是所有当前还在请求中的节点; 当请求完成,请求就会在这个队列里面移除; 如果这个队列一直是满的,说明当前负载很高或者网络连接有问题。如果所有 Node 都是满负载则请求不会被发起，除非等到队列数量减少。

    private final Map<String, Deque<NetworkClient.InFlightRequest>> requests = new HashMap<>();

复制代码

每个 Node 最大负载数 ？

每个客户端在发起请求还没有收到回复的时候都会被缓存到

InFlightRequests(空中请求数量)

里面,但是这个数量是有限制的,这个可以通过配置

max.in.flight.requests.per.connection

进行设置, 默认是: 5; 也就是每个客户端对每个 Node 最多也就同时发起 5 个未完成的请求; 如果超时这个数量就会等待有请求完成并释放额度了才可以发起新的请求；

相关的 Producer 配置有:

KeyValue 序列化

将 key 和 Value 先序列化。

自定义序列化器,需要实现

org.apache.kafka.common.serialization.Serializer

接口。我们简单看下

StringSerializer

序列化器

public class StringSerializer implements Serializer<String> {    private String encoding = "UTF8";    @Override    public void configure(Map<String, ?> configs, boolean isKey) {        String propertyName = isKey ? "key.serializer.encoding" : "value.serializer.encoding";        Object encodingValue = configs.get(propertyName);        if (encodingValue == null)            encodingValue = configs.get("serializer.encoding");        if (encodingValue instanceof String)            encoding = (String) encodingValue;    }    @Override    public byte[] serialize(String topic, String data) {        try {            if (data == null)                return null;            else                return data.getBytes(encoding);        } catch (UnsupportedEncodingException e) {            throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + encoding);        }    }}

复制代码

**configure(Map<String, ?> configs, boolean isKey)**这个方法是在构造

KafkaProduce

实例的时候调用的。

isKey

表示是 key 还是 value 来进行序列化这里 serialize(String topic, String data) 方法直接将字符串转换成 byte[]类型。

Kafka 客户端提供了很多种序列化器供我们选择,如果这些序列化器你都不满意,你也可以选择其他一些开源的序列化工具,或者自己进行实现。

计算分区号

将序列化后的 key、 value 调用合适的分区器选择将要发送的分区号。

分区三种策略

将消息缓存进 RecordAccumulator 累加器中

图解Kafka Producer中的消息缓存模型

Sender 发送消息

**Sender **线程在构造 KafkaProducer 的时候就已经启动了,它的职责就是从

以下忽略部分代码省略

    void runOnce() {        long currentTimeMs = time.milliseconds();        long pollTimeout = sendProducerData(currentTimeMs);        client.poll(pollTimeout, currentTimeMs);    }        private long sendProducerData(long now) {             // 获取哪些数据准备好了发送        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);    }

复制代码

寻找准备好发送的消息 Batch,获取对应 Leader 所在的 ReadyNode

我们都知道生产者生产的消息是暂时缓存在消息累加器 RecordAccumulator 中的, Sender 负责从 RecordAccumulator 里面获取准备好的数据进行发送

那么 ，哪些属于准备好的数据呢？

我们先回顾一下 RecordAccumulator 的结构。

在这里插入图片描述

每个 TopicPartition 的消息都会被暂存在 ProducerBatch Deque 阻塞队列中的其中一个 ProducerBatch 中,每个 ProducerBatch 都存放着一条或者多条消息。

具体请看 图解Kafka Producer 消息缓存模型

满足发送的条件的 Batch

遍历每个 TopicPartition 里面的 Deque, 获取队列中的第一个 ProducerBatch 如果该 TopicPartition 不存在 Leader,则忽略该 Batch,如果有则进入判断流程

因为消息是要发 Leader 所在的 Broker 发送的, 所以必须要有 Leader。

在满足条件

不属于重试或者属于重试并且等待的时候大于**

retry.backoff.ms

**** 的前提下,满足下面条件的均可发送**

（该条件就是要排除那些是属于重试,但是还没有到达重试间隔时间的情况。）

该 ProducerBatch 还没有被发送过. 该 Batch 能否发送判断条件如下

1. 如果该 Batch 满了或者 Batch 所在的 Deque 数量>1(数量大于 1 说明第一个 Batch 肯定就满了) 则满足发送条件

1. 如果消息累加器中内存用完了，有线程阻塞等待写入消息累加器 则也满足发送条件

1. RecordAccumulator 消息累加器被关闭,满足条件;（一般 KafkaProducer 被正常关闭的时候会先将累加器标记为已经关闭,方便让累加器里面的消息都发出去）
2. 是否被强制将消息发送出去。消息累加器 RecordAccumulator 提供强制**flush()方法供调用,用于该时刻的消息都满足发送的条件,一般在消息事务的地方有调用。 这里要注意的是,是调用flush()**这一时刻的所有未发送的 Batch 都需满足发送条件,后面新增的 Batch 不属于这一范畴
3. *该 Batch 的创建时间>***linger.ms**的时间

获取可发送请求的服务端 ReadyNodes

上面是讲哪些 Batch 属于可发送的逻辑判断,但是实际上,真正发送的时候并不是以每个 Batch 维度来判断发送的,而是以 Node 维度来发送的,上面我们知道了哪些 Batch 能够发送,然后我们就可以推断出 Batch 对应的 TopicPartition 所属的 Broker。有了这些可发送的 Broker,然后再来遍历 Broker 上的每个 TopicPartition 中的 First Batch

文字不好理解,我们看看下图

在这里插入图片描述

上图是生产者的 RecordAccumulator 消息累加器, 消息累加成上图所示。

每个 TopicPartition 队列都有很多 Batch, 我们知道了 TopicPartition 是不是就能够确定它所在的 Broker？

例如上图中

1. Topic1Partition-1、 Topic1Partition-2 、Topic2Partition-0 他们三个的 Leader 都存在于 Broker-0 中虽然 Topic2Partition-0 队列中不满足发送逻辑, 但是跟他同一个 Broker 中有其他的队列满足条件了,所以它最终也是满足发送条件的。
2. Topic2Partition-1 Leader 在 Broker-1 中,但是它不满足发送条件,这个 Broker 中也没有其他的满足条件了,所以客户端不会向 Broker-1 这个 Node 发起请求。
3. Topic1Partition-3 Leader 在 Broker-2 中,它满足发送条件,那么 Broker-2 就满足发送条件

那么最终得到的 ReadyNodes 就是 Broker-0、Broker-2

强制更新没有 Leader 的 Topic 元信息

上面我们在获取 哪些 Batch 准备好发送的时候,也会找到哪些 TopicPartition 没有 Leader。

那么这个时候就需要强制的去更新一下这些 TopicPartition 的元信息了,否则就发送不了。

过滤一些还未准备好连接的 ReadyNodes

上面我们已经获取了 ReadyNodes

那么在真正的向对应的 ReadyNodes 发起请求之前, 我们还是需要判断一下 我们的生产者客户端是否准备好了跟 ReadyNodes 发起请求.

那么客户端的准备条件有哪些呢？

生产者客户端在最开始的时候都没有跟任何 Node 建立连接的, 当我们尝试发送之前会去检验一下连接是否建立成功(就是当前这一步), 如果没有的话,则会去尝试建立连接。并且当前这次是会把这个 Node 过滤掉的,因为还没有建立成功链接,等到下一次循环的时候,可能已经建立成功了。

当然客户端是否准备好,不仅仅是判断 连接是否建立成功。

**还需要判断 当前未完成的请求队列数量是否 < ****

**max.in.flight.requests.per.connection**

**

遍历 ReadNodes 上的所有 TopicPartition 阻塞队列中的 FirstBatch 进行打包

到现在为止,我们已经得到了可以发送请求的 ReadyNodes 了。那么接下来就是分别解析这些 ReadyNode 他们能够发送的 Batch 打包发送了。

这一步最重要的作用是将 ProducerBatch 跟 Node 映射,也就是知道当前批次想哪个 Broker 发送哪些 Batch

   public Map<Integer, List<ProducerBatch>> drain(Cluster cluster, Set<Node> nodes, int maxSize, long now) {        if (nodes.isEmpty())            return Collections.emptyMap();  // 遍历ReadyNodes 每个Node下的队列都获取一遍        Map<Integer, List<ProducerBatch>> batches = new HashMap<>();        for (Node node : nodes) {            List<ProducerBatch> ready = drainBatchesForOneNode(cluster, node, maxSize, now);            batches.put(node.id(), ready);        }        return batches;    }

复制代码

那么应该选择哪些 Batch 来发送呢？

遍历每个 ReadyNode 节点下面的每个 TopicPartition 队列的首个 Batch

在这里插入图片描述

1. 如果 FirstBatch 属于重试, 并且还没有达到重试间隔时间retry.backoff.ms****, 则该 TopicPartition 队列会忽略 例如上图 Topic3Partition-1)
2. 如果 FirstBatch 为空, 则该 TopicPartition 队列会忽略；如左边 Topic3Partition-0
3. 如果该批次中的总 Batch 大小 > max.request.size 了. 则会终止此次遍历,并记录当前遍历到的位置, 等下次再次发送的时候从上一次结束的位置进行遍历 (但是这里 kafka 用了一个全局变量记录当前遍历到的索引,不是每个 Broker 一个变量, 是一个小 Bug)
4. 一次 Request 最多只会完整的遍历一遍, 就算遍历完一遍所有 TopicPartition 之后还没有写满max.request.size****. 那么也不会再重新遍历。

构造 Produce 请求并发起接着处理 Response

上面我们已经得到了

 Map<Integer, List<ProducerBatch>> batches

也就是 Node.id 和对应要发往该 Node 的 Request 请求携带的 ProducerBatch 列表。

发送成功之后,会返回 Response,根据 Response 情况处理不同的逻辑

Response 处理逻辑每个 Batch 都会对应着一个 PartitionResponse, 不同的 PartitionResponse 对应的不同处理逻辑。

1. 如果 Response 返回 RecordTooLargeException 异常,并且 Batch 里面的消息数量>1.这种情况, 就会尝试的去拆分 Batch, 如何拆分呢？ 是以**batch.size**大小来拆分成多个 Batch。并且重新放入到消息累加器中。
2. 如果返回是其他异常则先判断一下是否能够重试，如果能够重试,则重新入队到消息累加器中。重新入队的 Batch 会记录重试次数和时间等等信息。是否能够重试判断逻辑：batch 没有超过delivery.timeout.ms** && 重启次数<****retiries**
3. 如果是 DuplicateSequenceException 异常的话,那么并不会做其他的处理，而是当做正常完成。
4. 其他异常或者没有异常则会走正常流程, 并且调用 InterceptorCallback,如果有 Exception 也会返回。这个 InterceptorCallback 里面包含在拦截器 interceptors 和 userCallback(用户自己的回调)。调用顺序如下图：

1. 这个 usercallback 呢就是我们自己设置的，例如：** producer.send(record,new SzzTestCallBack(record.topic(), record.key(), record.value())); **注意: 这里的回调并不是指的一个 Batch 一个回调,这里是一个 Batch 里面有多少条消息,就有多少个回调。每个 ProducerBatch 里面都有一个对象专门保存所有消息的回调信息 thunks . 在处理 ProducerBatch 返回信息的时候会遍历这个 trunks, 来执行每个消息的回调。

假如你想确定某个消息是否发送成功, 那么你可以自己定义一个拦截器。并重写接口**

onAcknowledgement(RecordMetadata metadata, Exception exception)

**** 在这里面来判断你的消息是否发送成功。**

counter(counterh2)发送流程总结

Kafka Producer 整体架构图

整个生产者客户端是由主线程和 Sender 线程协调运行的, 主线程创建消息, 然后通过 拦截器、元信息更新、序列化、分区器、缓存消息等等流程。

然后 Sender 线程在初始化的时候就已经运行了,并且是一个 while 循环。

Sender 线程里面主要工作是：

1. 寻找 ReadyNodes: 去消息累加器里面获取有哪些 Node 是能够发送 Request 的。只要该 Node 有一个 TopicPartition 队列中有符合发送条件的 Batch。那么这个 Node 就应该是 ReadyNode。具体的筛选逻辑请看上文有具体分析。
2. 构建 Request: 过滤之后, 拿到了所有的 ReadyNodes。接下来就是遍历该 Node 下所有的 TopicPartition 队列里面的 FirstBatch, 组装到 Request 请求里面。发往一个 Node 的请求 Request,可以包含多个 ProducerBatch，能够一次发送多少个 Batch 是由配置max.request.size决定的，一个 Node 对应一个 Request。注意: 这个时候映射关系已经是 Map<Node,Reqeust>
3. 将 Request 放入 inFightRequest 中: 上面是组装好了 Request, 组装好了之后要先把这个 Request 放到inFightRequest对象中, 它保存着每个 Node 当前已经发送的 Request 但是还没有收到 Response 的请求。每个 Node 最多能够存放多少个未完成的 Request,是由max.in.flight.requests.per.connection控制的。需要注意的是, 如果队列已经满了, Request 是放入不了这个对象里面的,并且会抛出异常:"Attempt to send a request to node " + nodeId + " which is not ready."它决定着生产者针对某个 Node 的并发度。
4. Request 通过 Selector 发起通信.
5. 返回 Response： 服务端处理完成, 返回 Response 信息。
6. 从 inFightRequest 中移除完成 Request
7. 释放内存回消息累加器： 请求结束,清理消息累加器，将发送成功的 ProducerBatch 占用的内存大小加回到消息累加器中。注意：这里纯粹的是数字的加减,不涉及内存的处理, 因为发送成功之前的 Batch 占用了消息累加器的剩余可用内存。发送成之后要加回来。否则消息累加器满了会导致阻塞。