ToB企服应用市场:ToB评测及商务社交产业平台

标题: 【消息队列】Kafka从入门到口试学习总结 [打印本页]
作者: 知者何南    时间: 2024-10-14 18:50
标题: 【消息队列】Kafka从入门到口试学习总结
国科大学习生存（期末复习资料、课程大作业解析、大厂训练履历心得等）: 文章专栏（点击跳转）
大数据开发学习文档（分布式文件体系的实现，大数据生态圈学习文档等）: 文章专栏（点击跳转）
  
  
前言：Kafka最新版本已经可以不依赖ZooKeeper来实现，但经典永不外时，再怎么变其原理都是一致的。故本文基于依赖ZooKeeper的Kafka3.x展开。
1. 什么是Kafka？

1.1 Kafka定义

Kafka传统定义：Kafka是一个分布式的基于发布/订阅模式的消息队列（MessageQueue），主要应用于大数据实时处理范畴。

Kafka最新定义：Kafka是一个开源的分布式事件流平台（Event Streaming
Platform），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。

1.2 Kafka主要应用场景

缓冲/消峰：有助于控制和优化数据流经过体系的速率，解决生产消息和消费消息的处理速率不一致的情况。

解耦：答应你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

异步通讯：答应用户把一个消息放入队列，但并不立即处理它，然后在必要的时间再去处理它们。

1.3 Kafka根本架构



注：消息队列的两种模式：
点对点模式：仅有一个生产者一个消费者，消费者主动拉取数据，收到数据后立刻删除消息队列中数据。
发布/订阅模式：实用于多个生产者多个消费者的场景；消费者消费数据后不立即删除数据、每个消费者相互独立消费数据。
2. Kafka 生产者

2.1 生产者消息发送流程


流程图详解：
Interceptors：拦截器，Kafka通常不利用拦截器对数据举行处理。通常是通过上游数据源数据采集工具举行处理（如Flume）
Serializer：序列化器（因为要举行网络通讯）
Partitioner：分区器，因为Kafka是以分布式存储的方式存储数据
batch.size：只有数据积累到batch.size之后，sender才会发送数据。默认16k
linger.ms：假如数据迟迟未到达batch.size，sender等待linger.ms设置的时间到了之后就会发送数据。单位ms，默认值是0ms，表现没有延迟。
ACK应答：Kafka集群收到消息后的应答级别

总体流程：
在消息发送的过程中，涉及到了两个线程——Main线程和Sender线程。在main线程中创建了一个双端队列RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender 线程不停从RecordAccumulator中拉取消息发送到Kafka Broker。
2.2 生产者分区

2.2.1 分区的原因


2.2.2 分区计谋


2.3 生产者如何提高吞吐量

从上述生产者Producer发送数据到Kafka Broker的流程图可知，要想提高生产者吞吐量：
  1.         // batch.size：批次大小，默认16K
  2.     properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
  4.     // linger.ms：等待时间，默认0
  5.     properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
  7.     // RecordAccumulator：缓冲区大小，默认32M：buffer.memory
  8.     properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,                                               
  10.     // compression.type：压缩，默认none，可配置值gzip、snappy、lz4和zstd
  11.         properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");
复制代码
2.4 生产履历

2.4.1 数据可靠性


需求：正确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。 Kafka 0.11版本以后，引入了一项庞大特性。
2.4.2 幂等性和事务


Kafka生产者事务一共有如下5个API：
  1.         // 1初始化事务
  2.     void initTransactions();
  3.     // 2开启事务
  4.     void beginTransaction() throws ProducerFencedException;
  5.     // 3在事务内提交已经消费的偏移量（主要用于消费者）
  6.     void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,String consumerGroupId) throws ProducerFencedException;
  7.     // 4提交事务
  8.     void commitTransaction() throws ProducerFencedException;
  9.     // 5放弃事务（类似于回滚事务的操作）
  10.     void abortTransaction() throws ProducerFencedException;
复制代码
2.4.3 数据次序乱序问题

由生产者Producer发送数据流程图：

放大Sender线程部分：

kafka在1.x版本之前保证数据单分区有序，条件如下：
max.in.flight.requests.per.connection = 1（不必要思量是否开启幂等性）。即：将上图中的缓存请求数改为1个。
kafka在1.x及以后版本保证数据单分区有序，条件如下：
（1）未开启幂等性
max.in.flight.requests.per.connection必要设置为1。同样是将上图中的缓存请求数改为1个。
（2）开启幂等性
max.in.flight.requests.per.connection必要设置小于等于5。 故无论如何，都可以保证最近5个request的数据都是有序的。因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，
3. Kafka Broker

3.1 Kafka Broker工作流程

在zookeeper的服务端存储的Kafka相关信息：


总体流程：
AR：Kafka分区中的全部副本统称为AR（Assigned Repllicas）
ISR：表现和Leader保持同步的 Follower集合。
OSR：表现Follower与Leader副本同步时，延迟过多的副本。
AR = ISR + OSR
3.2 生产履历

3.2.1 服役退役节点

3.2.2 Leader 和 Follower 故障处理细节

LEO（Log End Offset）：每个副本的最后一个offset，LEO其实就是最新的offset + 1。
HW（High Watermark）：全部副本中最小的LEO 。
注意：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。
3.3 Kafka文件存储

3.3.1 文件存储机制

Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不停追加到该log文件末了（次序写），为防止log文件过大导致数据定位服从低下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的定名规则为：topic名称+分区序号，比方：first-0。

稀疏索引
.index文件为稀疏索引。约莫每往.log文件中写入4kb数据（默认，可以利用og.index.interval.bytes来调整），就会往.index文件写入一条索引。
.index文件中生存的offset为相对offset，如许能确保offset的值所占空间不会过大，因此能将offset的值控制在固定巨细。
查找数据过程
首先根据目标offset定位到segment文件；然后找到小于等于目标offset的最大offset所对应的索引项；定位到log文件；向下遍历找到目标Record。
3.3.2 文件清理计谋

Kafka 中默认的日记生存时间为7天，可以通过调整如下参数修改生存时间。
最低优先级小时，默认7天。 log.retention.hours，
分钟。log.retention.minutes，
最高优先级毫秒。log.retention.ms，
负责设置检查周期，默认5分钟。 log.retention.check.interval.ms，
当日记超过了设置的生存时间，会触发清理计谋：delete（删除）或compact（压缩）：

这种计谋只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩计谋，整个消息集里就生存了全部用户最新的资料。
3.4 高效读写数据


以读消费数据过程为例：

4. Kafka 消费者

4.1 Kafka 消费者工作流程


Consumer Group（CG）：消费者组，由多个consumer构成。形成一个消费者组的条件，是全部消费者的groupid相同。
消费者组内每个消费者负责消费差别分区的数据（可能是多个Topic），一个分区只能由一个组内消费者消费。
消费者组之间互不影响。全部的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
假如向消费组中添加更多的消费者，超过Topic分区数量，则有一部分消费者就会闲置，不会接收任何消息。
4.2 消费者组初始化流程

Coordinator：辅助实现消费者组的初始化和分区的分配。
Coordinator节点选择 = Groupid的Hashcode值% 50（ __consumer_offsets的分区数量）
比方：Groupid的Hashcode值 = 1，1 % 50 = 1，那么__consumer_offsets主题的1号分区，在哪个broker上，就选择这个节点的Coordinator作为这个消费者组的老大。消费者组下的全部的消费者提交offset的时间就往这个分区去提交offset。

总体流程：
当Consumer启动后会向Coordinator发起JoinGroup请求，此时Coordinator会选出一个Consumer作为Leader，并将要消费的Topic情况发送给Leader消费者。Leader消费者收到Topic信息后会制定消费方案并将消费方案发给Coordinator。此时Coordinator会将该消费方案广播到各个Follower消费者。然后各消费者开始消费。

4.3 消费者组详细消费流程


总体流程：
消费者组中的消费者发起消费数据请求sendFetches，Kafka集群中的数据分区将数据发送到completedFetches队列中，然后消费者通过FetchedRecordes从队列中拉取数据，然后依次经过反序列化器、拦截器后得到数据。
Fetch.min.bytes每批次最小抓取巨细，默认1字节
fetch.max.wait.ms一批数据最小值未到达的超时时间，默认500ms
Fetch.max.bytes每批次最大抓取巨细，默认50m
FetchedRecords请求从队列中抓取数据
Max.poll.records一次拉取数据返回消息的 最大条数，默认500条
独立消费者消费数据方式：（订阅主题、（订阅分区
消费者组消费数据方式：同一个主题的分区数据，只能由一个消费者组中的一个消费。
4.4 分区分配及再均衡计谋

一个consumer group中有多个consumer构成，一个topic有多个partition构成，现在的问题是，到底由哪个consumer来消费哪个partition的数据？
Kafka有四种主流的分区分配计谋：Range、RoundRobin、Sticky、CooperativeSticky。
可以通过配置参数partition.assignment.strategy，修改分区的分配计谋。默认计谋是Range + CooperativeSticky。Kafka可以同时利用多个分区分配计谋。
分区分配计谋之 Range
分区分配计谋
Range是对每个Topic而言的。
举例：当前有七个分区，三个消费者，排序后的分区为0.1.2.3.4.5.6.7；消费者排序完之后为C1 C2 C3.那么8/3 = 2 余 2，那么C1消费0.1.2三个分区、C2消费3.4.5三个分区，C3消费6.7两个分区。
注意：此计谋是针对单个Topic而言的，假如集群中有N多个topic，那么消费者C1会比其他消费者多消费大量分区，容易造成数据倾斜。
再均衡计谋

分区分配计谋之 RoundRobin
分区分配计谋
RoundRobin是针对集群内全部Topic而言的。
RoundRobin 轮询分区计谋，是把全部的 partition 和全部的consumer 都列出来；
然后按照hashcode举行排序，最后通过轮询算法来分配 partition 给到各个消费者。
再均衡计谋

分区分配计谋之 Sticky
分区分配计谋
粘性分区定义：可以明白为分配的结果带有“粘性的”。即在实验一次新的分配之前，思量上一次分配的结果，只管少的调整分配的变动，可以节省大量的开销。
粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配计谋，首先会只管均衡的放置分区到消费者上面；在出现同一消费者组内消费者出现问题的时间，会只管保持原有分配的分区稳定化。
再均衡计谋

4.5 offset 位移

Kafka0.9版本之前，consumer默认将offset生存在Zookeeper中；从0.9版本开始，consumer默认将offset生存在Kafka 一个内置的topic中，该topic为__consumer_offsets。
__consumer_offsets 主题里面接纳 key 和 value的方式存储数据。key是 group.id+topic+分区号，value 就是当前 offset 的值。每隔一段时间，kafka 内部会对这个 topic 举行Compact，也就是每个group.id + topic + 分区号就保留最新数据。
4.5.1 offset提交计谋


4.5.2 漏消费和重复消费


重复消费：已经消费了数据，但是offset没提交。

漏消费：先提交offset后消费，有可能会造成数据的漏消费。

怎么能做到既不漏消费也不重复消费呢？详看消费者事务。
4.6 生产履历

4.6.1 消费者事务

假如想完成Consumer端的精准一次性消费，那么必要Kafka消费端将消费过程和提交offset过程做原子绑定。此时我们必要将Kafka的offset生存到支持事务的自定义介质（比如MySQL）。这部分知识会在后续项目部分涉及。

4.6.2 数据积压（消费者如何提高吞吐量）

5. Kafka-Eagle监控

Kafka-Eagle框架可以监控Kafka集群的整体运行情况，在生产情况中常常利用。
焦点原理和功能：
思考：假如不利用Kafka-Eagle如何实现监控Kafka集群功能？
Interceptor拦截器！可以通过在此处设置检查点，同样可以实现监控Kafka集群的功能。
6. Kafka-Kraft模式

6.1 基于ZK与基于KRaft的Kafka

Zookeeper 提供了配置服务、分布式同步、定名服务、Leader 选举和集群管理等功能，在大数据期间的开始很多开源产品都依赖 Zookeeper 来构建，Apache Kafka 也不例外。但是随着 Kafka 功能的演进和应用的场景越来越多：

因此作为 Zookeeper 的替代，Kafka 3.3.1 提供了 KRaft 元数据管理组件。下图来自于 KIP-500 [1]提案，左右分别是 Zookeeper 模式和 KRaft 模式的摆设架构图。

6.2 新旧模式对比


ZAB协议以及Raft协议先容可以参考我别的一篇文章：
分布式一致性算法Paxos、Raft 及 Zookeeper ZAB 蹒跚者_Stan_-CSDN博客
7. Kafka外部体系集成

集成Flume、Spark、Flink
Flume、Spark、Flink是在大数据开发范畴中非常常用的组件。他们都既可以用于Kafka的生产者，也可以用于 Kafka的消费者。

详细实现参考我的其他文章：
待增补
参考文献

Kafka 如何基于 KRaft 实现集群最终一致性和谐 - SegmentFault 思否
Kafka3.x教程（从入门到调优，深入全面）_哔哩哔哩_bilibili
KIP-500: Replace ZooKeeper with a Self-Managed Metadata Quorum - Apache Kafka - Apache
  Software Foundation
项目地址

BigDataDev: 大数据焦点框架学习pro (gitee.com)

接待各人参考！
   14点08分 2024年10月13日
消息队列 Kafka内容学习整理，如有错误，接待评论区交换指出。
不积跬步无以至千里！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4