参考023 - 大数据 - Kafka - 生产者 - 生产数据的准备_哔哩哔哩_bilibili

链接：https://pan.baidu.com/s/1QMOJVkRy4nKkjzoDryvQXw
提取码：fcoe

本文接着上一篇kafka和flink的入门到精通 3 组件扩展，kafka-生产者_水w的博客-CSDN博客

4.1 生产数据流程

◼ 生产数据的准备

➢ KafkaProducer，ProducerRecod

◼ 采集器

◼ 拦截器的实现

◼ 元数据请求和更新

◼ 分区选择

◼ 将数据缓存到采集器中

◼ sender从采集器中获取数据发送到服务器

4.1 生产数据流程

◼ 生产数据的准备

➢ KafkaProducer，ProducerRecod

这个“KafkaProducer”下面是一些初始化的配置操作，

clientId：；
partitioner：是可配置的，默认为DefaultPartitioner；
interceptors：拦截器；
maxRequestSize：最大请求大小；
compressionType：压缩方式；
accumulator：采集器；
metadata：元数据；
sender：发送对象，会随着线程的启动而执行；
ioThread；

因此，accumulator就是快递员-----缓冲区，sender就是运输工具。

ProducerRecord的参数：

◼ 采集器

小人要发送快递，做的第一件事就是 send操作来发送数据，把record做了一系列的拦截interceptors（不是只有一个），做了一些定制化的操作，形成了新的Record。

当把消息拦截之后，那么把消息发给谁，我们不知道。那么这些东西在哪，就在元数据当中。需要从元数据当中得到一些信息，比如集群的信息。然后把Key做序列化，消息继续往下走。

但是Topic主题有多个分区，那么接下来就要问消息要放在哪个分区Partition？因为如果都往一个分区里放会导致过载。-------分区选择

现在Topic和分区Partition我都知道了，那么TopicPartition对象就会被构建出来，构建出来之后，开始做一些检验操作，会判断当前数据的大小有没有超过阈值。

接下来，重点是一个RecordAccumulator缓冲区，快递就被快递员拿到了。

这就是一个比较完整的流程了。那么接下来就详细说说。

◼ 拦截器的实现

生成一个自定义拦截器“MyProducerInterceptor”，继承“ProducerInterceptor”，在新的拦截器中定义onSend()方法，为了好区分，我们让key和Value不一样。（原本的拦截器中的onSend()方法，key和Value是一样的。）

将生成的自定义拦截器“MyProducerInterceptor”进行配置，

运行代码，刷新之后，会发现多了一条数据，

这就叫拦截器。

◼ 元数据请求和更新

此时，拦截器已经处理完了，

Matadata不是频繁的从服务器得到数据，它是可以缓存下来的，缓存以后做一些事情。

可以看到，这儿的代码“watiOnMetadata” 等待元数据，那么为什么要等待元数据？因为元数据中包含很多信息和内容，我们需要知道里面有什么。

cache：缓存；
newTopics
cluster：集群；

那么集群中有什么？

可以看到， cluster集群中有：

nodes：服务器节点的集合；
controller；
partitionByTopicPartition：当亲的分区里的分区信息；
partitionByTopic：某一个主题topic当中的所有分区；
availablePartitionByTopic：可用的分区；
partitionByNode：某一个node节点当中的所有分区；

等等，里面东西太多了，这里不一一列举了。

剩下的，我们就需要怎么去等待元数据的？

那么接下来，我们就需要知道是怎么去等待元数据的？我们怎么就把元数据得到了呢？

默认情况下是肯定不知道元数据的，因为最开始的时候，第一次发送数据，我们都不知道服务器有哪些分区，那么我们要怎么划分分区？连元数据都没有，我们怎么做分区选择？

所以首先要把元数据得到。

可以看到，requestUpdateForTopic(topic) 就意味着我准备发送请求来更新元数据了。

可以看到代码中，sender唤醒，进行awaitUpdate()来等待更新，什么时候更新完毕了，程序才会继续走下去。所以现在这个awaitUpdate()就会阻塞等待，sender是一个线程，sender.wakeup()就会执行，那么也就是说当前的生产线程（主线程）就会暂停，seder发送线程就会开始。

比如，我现在想发到广州一个快递，问快递员说能不能发到广州，快递员说你等一等我问一下。他得知道它们公司能不能发到广州。问了公司说可以发，快递员才会给你说可以，我就得到了这个信息。

sender会向集群的服务器Kafka Servers发送一个请求MatadataRequest(METADATA)，传了一个ApiKeys叫METADATA；
doSend()向服务器发送request请求；
请求发送到了Kafka Servers之后，放到了requestChnnel中，然后再从requestChnnel中把request请求取出来，KafkaApis会得到请求对象。
KafkaApis中， 到了handle这一步，就会取到请求对象了，拿到了ApiKey；
得到以后，到了handleTopicMetaRequest来处理这个请求，拿到了请求信息，拿到之后就可以把所有的Topic和分区都可以得到了。得到以后，发给客户端。
此时，sender得到响应对象response，对响应处理，其中如果是metadataResonse，就会有一个处理成功的响应handleSuccessfulResonse。就完成了设定，进行更新update，当前就有数据了。