Producer源码解读

Producer源码解读

在 Kafka 中, 我们把产生消息的一方称为 Producer 即 生产者, 它是 Kafka 的核心组件之一, 也是消息的来源所在。它的主要功能是将客户端的请求打包封装发送到 kafka 集群的某个 Topic 的某个分区上。那么这些生产者产生的消息是怎么传到 Kafka 服务端的呢？

Producer之整体流程

image.png

但是站在源码的核心角度，我们可以把Producer分成以下几个核心部分：

1、Producer之初始化

2、Producer之发送流程

3、Producer之缓冲区

4、Producer之参数与调优

Producer源码解读

从生产流程可以知道，Producer里面的核心有序列化器，分区器，还有缓冲，所以初始化的流程肯定是围绕这几个核心来处理。

image.png

KafkaProducer之初始化

image.png

image.png

1)、设置分区器

设置分区器(partitioner)，分区器是支持自定义的

image.png

2)、设置每次重试间隔时间

设置每次重试间隔时间(retryBackoffMs)默认100ms

如果发送消息到broker时抛出异常，且是允许重试的异常，那么就会最大重试retries参数指定的次数，同时retryBackoffMs是重试的间隔。

image.png

3)、设置序列化器

设置序列化器(Serializer)

image.png

4)、设置拦截器

设置拦截器(interceptors)，关于拦截器，这个后面会有讲解和介绍。

image.png

5)、设置缓冲区

image.png

在之前，还有一些参数的设置。

image.png

1、设置最大的消息为多大(maxRequestSize), 默认最大1M, 生产环境可以提高到10M

2、设置缓存大小(totalMemorySize) 默认是32M

3、设置压缩格式(compressionType)

4、初始化RecordAccumulator也就是缓冲区指定为32M

6)、设置消息累加器

因为生产者是通过缓冲的方式发送，发送的条件之前的课程讲过，所以这里需要一个消息累加器配合才能完成消息的发送。

image.png

5、初始化集群元数据(metadata),刚开始空的

image.png

6)、创建Sender线程

image.png

这里还初始化了一个重要的管理网路的组件 NetworkClient

image.png

KafkaThread将Sender设置为守护线程并启动

image.png

拦截器使用及介绍

这里讲一讲拦截器的使用和基本作用，拦截器一般用得不多，所以这里只是讲一讲案例，不推荐生产中使用。

想要实现拦截器，我们需要先实现ProducerInterceptor接口即可，然后在生产者中设置进去即可。

image.png

image.png

1、想要把发送的数据都带上时间戳image.png

2、实现统计发送消息的成功次数和失败次数

在 onAcknowledgement(RecordMetadata, Exception)里面，根据消息发送后返回的异常信息来判断是否发送成功。一般异常如果为空就说明发送成功了，反之就说明发送失败了。

然后定义两个变量，并根据Exception的值分别累加就可以统计到了

最后在close方法里输出两个变量的值，这样当producer发送数据结束并close后，会自动调用拦截器的close方法来输出咱们想要统计的成功和失败次数

image.png

image.png

不过这里要注意一个点：

onAcknowledgement运行在producer的IO线程中，因此不要在该方法中放入很复杂的逻辑，否则会拖慢producer的消息发送效率。

3、拦截链路

   // 设置属性Properties properties = new Properties();// 指定连接的kafka服务器的地址properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");// 设置String的序列化properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);//设置自定义拦截器properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, SelfInterceptor.class);//设置拦截链路(设置多个 SelfInterceptor  先执行   再执行SelfInterceptor2)ArrayList<String> interceptors = new ArrayList<>();interceptors.add("com.llp.interceptor.SelfInterceptor");//注意：这里是拦截器的全类名interceptors.add("com.llp.interceptor.SelfInterceptor2"); //这里假设有SelfInterceptor2properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);

拦截器链里的拦截器是按照顺序组成的，因此我们要注意前后拦截器对彼此的影响，比如这里拦截器1的onsend方法不能返回null，不然拦截器2的onsend就丢失了信息，会发生异常。

Producer之发送流程

Producer之发送流程

Kafka Producer 发送消息流程如下:

1)、执行拦截器逻辑

执行拦截器逻辑，预处理消息, 封装 Producer Record

image.png

2)、集群元数据

从 Kafka Broker 集群获取集群元数据metadata

image.png

3)、序列化

调用Serializer.serialize()方法进行消息的key/value序列化

image.png

4)、分区

调用partition()选择合适的分区策略，给消息体 Producer Record 分配要发送的 topic 分区号

image.png

5)、消息累加进缓存

将消息缓存到RecordAccumulator 收集器中, 最后判断是否要发送。

image.png

7)、消息发送

前面我们也知道真正的消息发送是Sender线程来做，并且这里还要结合缓冲区来处理。

批次发送的条件为:缓冲区数据大小达到 batch.size 或者 linger.ms 达到上限，哪个先达到就算哪个

Producer之缓冲区

Kafka生产者的缓冲区，也就是内存池，可以将其类比为连接池(DB, Redis),主要是避免不必要的创建连接的开销, 这样内存池可以对 RecordBatch 做到反复利用, 防止引起Full GC问题。那我们看看 Kafka 内存池是怎么设计的。

核心就是这段代码：

image.png

image.png

   Kafka 内存设计有两部分，下面的粉色的是可用的内存（未分配的内存，初始的时候是 32M），上面紫色的是已经被分配了的内存，每个小 Batch 是 16K，然后这一个个的 Batch 就可以被反复利用，不需要每次都申请内存,  两部分加起来是 32M。
申请内存的过程

从 Producer 发送流程的第6步中可以看到会把消息放入 accumulator中, 即调用 accumulator.append() 追加, 然后把消息封装成一个个Batch 进行发送, 然后去申请内存(free.allocate())

image.png

image.png

（1）如果申请的内存大小超过了整个缓存池的大小，则抛异常出来

image.png

（2）对整个方法加锁：

this.lock.lock();

（3）如果申请的大小是每个 recordBatch 的大小（16K），并且已分配内存不为空，则直接取出来一个返回。

if (size == poolableSize && !this.free.isEmpty())return this.free.pollFirst();

image.png

（4）如果整个内存池大小比要申请的内存大小大 （this.availableMemory + freeListSize >= size），则直接从可用内存（即上图粉色的区域）申请一块内存。并且可用内存要去掉申请的那一块内存。

image.png

Sender线程

image.png

Producer之参数调优

Kafka 实际使用中，Producer 端既要保证吞吐量，又要确保无消息丢失，一些核心参数的配置就显得至关重要。接下来我们就来看看生产端都有哪些重要的参数,及调优建议。

acks

参数说明：对于 Kafka Producer 来说是一个非常重要的参数，它表示指定分区中成功写入消息的副本数量，是 Kafka 生产端消息的持久性的保证, 详细可以查看

发送确认机制 3 种不同的确认模式。

acks=0 意味着如果生产者能够通过网络把消息发送出去，那么就认为消息已成功写入Kafka 。
acks=1 意味若首领在收到消息并把它写入到分区数据文件（不一定同步到磁盘上）时会返回确认或错误响应。
acks=all 意味着首领在返回确认或错误响应之前，会等待（min.insync.replicas）同步副本都收到悄息。

max.request.size

参数说明：这个参数对于 Kafka Producer 也比较重要， 表示生产端能够发送的最大消息大小，默认值为1048576(1M) 。

  调优建议：这个配置对于生产环境来说有点小， **为了避免因消息过大导致发送失败，生产环境建议适当调大，比如可以调到10485760(10M)** 。

retries

参数说明：表示生产端消息发送失败时的重试次数，默认值为0，即不重试。 这个参数一般是为了解决因系统瞬时故障导致的消息发送失败，比如网络抖动、Leader 选举及重选举，其中瞬时的 Leader 重选举是比较常见的。因此这个参数的设置对于 Kafka Producer 就显得非常重要 。

 调优建议：这里建议设置为一个大于0的值，比如3次。

retry.backoff.ms

参数说明：**设定两次重试之间的时间间隔，避免无效的频繁重试，默认值为100, 主要跟 retries 配合使用， **在配置 retries 和 retry.backoff.ms 之前，最好先估算一下可能的异常恢复时间，需要设定总的重试时间要大于异常恢复时间，避免生产者过早的放弃重试。

connections.max.idele.ms

参数说明：主要用来判断多久之后关闭空闲的链接，默认值540000（ms）即9分钟。

compression.type

参数说明： 该参数表示生产端是否要对消息进行压缩，默认值为不压缩(none)。 压缩可以显著减少网络IO传输、磁盘IO以及磁盘空间，从而提升整体吞吐量，但也是以牺牲CPU开销为代价的。

 调优建议：出于提升吞吐量的考虑，建议在生产端对消息进行压缩。**对于Kafka来说，综合考虑吞吐量与压缩比，建议选择lz4压缩。如果追求最高的压缩比则推荐zstd压缩。**

buffer.memory

参数说明： 该参数表示生产端消息缓冲池或缓冲区的大小，默认值为即33554432(32M) 。这个参数基本可以认为是 Producer 程序所使用的内存大小。

调优建议：通常我们应尽量保证生产端整体吞吐量，建议适当调大该参数，也意味着生产客户端会占用更多的内存。

batch.size

参数说明： 该参数表示发送到缓冲区中的消息会被封装成一个一个的Batch，分批次的发送到 Broker 端，默认值为16KB。 因此减小 batch 大小有利于降低消息延时，增加 batch 大小有利于提升吞吐量。

 调优建议：通常合理调大该参数值，能够显著提升生产端吞吐量，比如可以调整到32KB，调大也意味着消息会有相对较大的延时。

linger.ms

参数说明： 该参数表示用来控制 Batch 最大的空闲时间，超过该时间的 Batch 也会自动被发送到 Broker 端。 实际情况中, 这是吞吐量与延时之间的权衡。默认值为0，表示消息需要被立即发送，无需关系 batch 是否被填满。

  调优建议：通常为了减少请求次数、提升整体吞吐量，建议设置一个大于0的值，比如设置为100，此时会在负载低的情况下带来100ms的延时。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/241599.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

python数字图像处理基础（九）——特征匹配

python数字图像处理基础（九）——特征匹配

目录 蛮力匹配&#xff08;ORB匹配&#xff09;RANSAC算法全景图像拼接 蛮力匹配&#xff08;ORB匹配&#xff09; Brute-Force匹配非常简单&#xff0c;首先在第一幅图像中选取一个关键点然后依次与第二幅图像的每个关键点进行&#xff08;描述符&#xff09;距离测试&#x…
阅读更多...
2024.1.16 用户画像day01 - 项目介绍

2024.1.16 用户画像day01 - 项目介绍

目录 一. 项目介绍 整体流程: 项目建设目的: 学习安排: 技术选型: 技术架构: 项目架构: 二 . 名词解释 一. 项目介绍 整体流程: 项目介绍-elasticSearch-业务数据源导入-离线指标开发-Flume实时采集-Nginx日志埋点数据- 结构化流实时指标 - 制作报表 数仓开发用户画像…
阅读更多...
【Linux】基本指令

【Linux】基本指令

&#x1f525;博客主页&#xff1a; 小羊失眠啦. &#x1f3a5;系列专栏&#xff1a;《C语言》 《数据结构》 《Linux》《Cpolar》 ❤️感谢大家点赞&#x1f44d;收藏⭐评论✍️ 文章目录 一、认识操作系统二、Linux基本指令&#x1f4d6;2.1 ls指令&#x1f4d6;2.2 pwd指令&…
阅读更多...
利用浏览器开发者工具进行网页性能优化

利用浏览器开发者工具进行网页性能优化

目录 学习目标&#xff1a; 学习内容&#xff1a; 学习时间&#xff1a; 学习产出&#xff1a; 网页性能优化的基本概念和指标&#xff1a; 浏览器开发者工具的基本功能和使用方法&#xff1a; 使用网络面板进行网页加载性能分析&#xff1a; 使用性能面板进行网页渲染性能分析…
阅读更多...
SQL Povit函数使用及实例

SQL Povit函数使用及实例

PIVOT函数常用于数据的行转列&#xff0c;同时也可以用此函数实现类似于Excel中的数据透视表的效果。 PIVOT函数 PIVOT 函数的基本语法如下&#xff1a; -- PIVOT 语法 SELECT <非透视的列>,[第一个透视的列] AS <列名称>,[第二个透视的列] AS <列名称>,.…
阅读更多...
表的增删改查 进阶（二）

表的增删改查 进阶（二）

&#x1f3a5; 个人主页&#xff1a;Dikz12&#x1f525;个人专栏&#xff1a;MySql&#x1f4d5;格言&#xff1a;那些在暗处执拗生长的花&#xff0c;终有一日会馥郁传香欢迎大家&#x1f44d;点赞✍评论⭐收藏 目录 3.新增 4.查询 聚合查询 聚合函数 GROUP BY子句 HA…
阅读更多...
[论文阅读]DeepFusion

[论文阅读]DeepFusion

DeepFusion Lidar-Camera Deep Fusion for Multi-Modal 3D Object Detection 用于多模态 3D 物体检测的激光雷达相机深度融合 论文网址&#xff1a;DeepFusion 论文代码&#xff1a;DeepFusion 摘要 激光雷达和摄像头是关键传感器&#xff0c;可为自动驾驶中的 3D 检测提供补…
阅读更多...
嵌入式linux_C应用学习之API函数

嵌入式linux_C应用学习之API函数

1.文件IO 1.1 open打开文件 #include <sys/types.h> #include <sys/stat.h> #include <fcntl.h> int open(const char *pathname, int flags); int open(const char *pathname, int flags, mode_t mode);pathname&#xff1a;字符串类型&#xff0c;用于标…
阅读更多...
Node.js基础知识点（五）- http

Node.js基础知识点（五）- http

一.request 请求事件处理函数 var http require(http)var server http.createServer() request 请求事件处理函数&#xff0c;需要接收两个参数&#xff1a; Request 请求对象 请求对象可以用来获取客户端的一些请求信息&#xff0c;例如请求路径 Response 响应对象 响应对…
阅读更多...
泥石流识别摄像机

泥石流识别摄像机

泥石流是一种自然灾害&#xff0c;经常给人们的生命和财产带来严重威胁。因此&#xff0c;研发泥石流识别摄像机成为一种急迫需求。这类摄像机的主要功能是实时监测泥石流的形成和运动&#xff0c;及早发现泥石流的迹象&#xff0c;从而及时采取应急措施。 泥石流识别摄像机的设…
阅读更多...
lenovo联想笔记本电脑拯救者Legion Y7000 2019 PG0(81T0)原装出厂Windows10系统

lenovo联想笔记本电脑拯救者Legion Y7000 2019 PG0(81T0)原装出厂Windows10系统

链接&#xff1a;https://pan.baidu.com/s/1fn0aStc4sfAfgyOKtMiCCA?pwdas1l 提取码&#xff1a;as1l 联想拯救者原厂Win10系统自带所有驱动、出厂主题壁纸、系统属性联机支持标志、系统属性专属LOGO标志、Office办公软件、联想电脑管家等预装程序 所需要工具&#xff1a;…
阅读更多...
Linux上新部署的项目jar包没有生效

Linux上新部署的项目jar包没有生效

今天公司新安排了一个项目&#xff0c;这里简称项目A&#xff0c;需要新增两个功能&#xff0c;我这边完成之后&#xff0c;跟前端对接好了&#xff0c;调试也没有问题。 然后把项目打包上传到测试服务器上&#xff0c;重新启动项目&#xff0c;发现项目A新增的接口没有生效&a…
阅读更多...
【办公类-21-01】20240117育婴员操作题word合并1.0

【办公类-21-01】20240117育婴员操作题word合并1.0

背景需求&#xff1a; 最近学校组织老师们学习“育婴员”高级&#xff0c;每周学习2题操作&#xff0c;所以我是把每个学习内容单独做在一个word文件里 上周8套保健操作学完了&#xff0c;需要整理&#xff0c;并将8份Word文件合并 第一步&#xff1a;doc装docx 合并时程序报…
阅读更多...
QT上位机开发（MFC vs QT）

QT上位机开发（MFC vs QT）

【 声明：版权所有，欢迎转载，请勿用于商业用途。 联系信箱：feixiaoxing @163.com】 在qt之前，上位机开发的主要方法就是mfc。后来出现了c#语言之后，上位机的开发就有一部分人转成了c#。这些开发都是在windows平台完成的，而linux上面的界面，则都是通过各种小众库…
阅读更多...
深度学习中Numpy的一些注意点（多维数组；数据类型转换、数组扁平化、np.where()、np.argmax()、图像拼接、生成同shape的图片）

深度学习中Numpy的一些注意点（多维数组；数据类型转换、数组扁平化、np.where()、np.argmax()、图像拼接、生成同shape的图片）

文章目录 1多维数组压缩维度扩充维度 2numpy类型转换深度学习常见的float32类型。 3数组扁平化4np.where()的用法5np.argmax()6图像拼接7生成同shape的图片&#xff0c;指定数据类型 1多维数组 a.shape(3,2);既数组h3&#xff0c;w2 a.shape(2,3,2);这里第一个2表示axis0维度上…
阅读更多...
C++后端笔记

C++后端笔记

C后端笔记 资源整理一、高级语言程序设计1.1 进制1.2 程序结构基本知识1.3 数据类型ASCII码命名规则变量间的赋值浮点型变量的作用字符变量常变量 const运算符 二、高级语言程序设计&#xff08;荣&#xff09; 资源整理 C后端开发学习路线及推荐学习时间 C基础知识大全 C那…
阅读更多...
SPI 动态服务发现机制

SPI 动态服务发现机制

SPI&#xff08;Service Provier Interface&#xff09;是一种服务发现机制&#xff0c;通过ClassPath下的META—INF/services文件查找文件&#xff0c;自动加载文件中定义的类&#xff0c;再调用forName加载&#xff1b; spi可以很灵活的让接口和实现分离&#xff0c; 让API提…
阅读更多...
『MySQL快速上手』-⑩-索引特性

『MySQL快速上手』-⑩-索引特性

文章目录 1.索引的作用2.索引的理解建立测试表插入多条记录查看结果 2.1 MySQL与磁盘交互的基本单位2.1 为何IO交互要是 Page2.3 理解单个Page2.4 理解多个Page2.5 页目录2.6 单页情况2.7 多页情况2.8 B vs B2.9 聚簇索引 vs 非聚簇索引非聚簇索引聚簇索引 3.索引操作3.1 创建主…
阅读更多...
冻结Prompt微调LM： T5 PET （a）

冻结Prompt微调LM： T5 PET （a）

T5 paper: 2019.10 Exploring the Limits of Transfer Learning with a Unified Text-to-Text Transformer Task: Everything Prompt: 前缀式人工prompt Model: Encoder-Decoder Take Away: 加入前缀Prompt&#xff0c;所有NLP任务都可以转化为文本生成任务 T5论文的初衷如…
阅读更多...
macOS安装VMware Fusion 13试用版本

macOS安装VMware Fusion 13试用版本

1.下载: Download VMware Fusion | VMware 保存到桌面 下载成功: 双击dmg文件运行安装 安装成功 <
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023