【消息队列】Kafka从入门到口试学习总结

国科大学习生存（期末复习资料、课程大作业解析、大厂训练履历心得等）: 文章专栏（点击跳转）
大数据开发学习文档（分布式文件体系的实现，大数据生态圈学习文档等）: 文章专栏（点击跳转）
  
  
前言：Kafka最新版本已经可以不依赖ZooKeeper来实现，但经典永不外时，再怎么变其原理都是一致的。故本文基于依赖ZooKeeper的Kafka3.x展开。
1. 什么是Kafka？

1.1 Kafka定义

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

• 发布/订阅：消息的发布者不会将消息直接发送给特定的订阅者，而是将发布的消息分为差别的种别，订阅者只接收感兴趣的消息。
  

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

1.2 Kafka主要应用场景

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

• 比方：双十一期间，上有产生数据的速率为10亿/s，但是下游体系消费数据的速率为1千万/s，此时就必要一个消息队列来起到缓冲消峰的作用。
  

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

• 比方：假如遇到上游有多个生产者数据源、下游有多个消费者时，相互之间获取数据的时间就必要遵守各自的接口规范，导致体系之间的耦合度较高。此时就必要消息队列来对他们举行解耦，生产者与消费者仅需和Kafka打交道而无需关注多变的接口约束。
  

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

• 比方：用户注册某账号后必要发送注册成功的短信，假如注册成功的消息需等待短信发送成功后再返回则会等待较长时间，此时可以利用Kafka来做异步通讯：用户注册成功后直接返回注册成功，并将发送短信的请求放入Kafka中，由下游发送短信接口次序消费即可。
  

1.3 Kafka根本架构

• Topic：可以明白为一个队列，生产者和消费者面向的都是一个topic。
  
• Broker：一台 Kafka服务器就是一个broker。一个集群由多个broker构成。一个broker 可以容纳多个topic。
  
• Producer：消息生产者，就是向Kafka broker发消息的客户端。
  
• Consumer：消息消费者，向Kafka broker取消息的客户端。
  
• Consumer Group（CG）： 消费者组，由多个consumer构成。消费者组内每个消费者负责消费差别分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。全部的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
  
• Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服务器）上，一个topic可以分为多个partition（物理上的概念），每个partition是一个有序的队列。
  
• Replica：副本。一个 topic 的每个分区都有若干个副本，一个 Leader 和若干个Follower。
  
• Leader：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是Leader。
  
• Follower：每个分区多个副本中的“从”，实时从 Leader 中同步数据，保持和Leader 数据的同步。Leader发生故障时，某个Follower会成为新的Leader。
  

注：消息队列的两种模式：
点对点模式：仅有一个生产者一个消费者，消费者主动拉取数据，收到数据后立刻删除消息队列中数据。
发布/订阅模式：实用于多个生产者多个消费者的场景；消费者消费数据后不立即删除数据、每个消费者相互独立消费数据。
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集群收到消息后的应答级别

• 0：生产者发送过来的数据，不必要等数据落盘应答。
  
• 1：生产者发送过来的数据，Leader收到数据后应答。
  
• -1（all）：生产者发送过来的数据，Leader和ISR队列里面的全部节点收齐数据后应答。-1和all等价。
  

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

2.2.1 分区的原因

• 提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位举行消费数据。
  
• 便于公道利用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。公道控制分区的任务，可以实现负载均衡的结果。
  
  
  
  

2.2.2 分区计谋

• 指明partition的情况下，直接将指明的值作为partition值；
  
  
  
  • 比方：partition=0，全部数据写入分区0。
    
  
  
• 没有指明partition值但有key的情况下，将key的hash值与topic的partition数举行取余得到partition值；
  
  
  
  • 比方：key1的hash值=5，key2的hash值=6 ，topic的partition数=2，那么key1 对应的value1写入1号分区，key2对应的value2写入0号分区。
    
  
  
• 既没有partition值又没有key值的情况下，Kafka接纳Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直利用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区举行利用（和上一次的分区差别）。
  
  
  
  • 比方：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到，Kafka再随机一个分区举行利用（假如照旧0会继承随机）。
    
  
  

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 数据可靠性

• 至少一次（AtLeastOnce）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2；可以保证数据不丢失，但是不能保证数据不重复。
  
• 最多一次（AtMostOnce）= ACK级别设置为0；可以保证数据不重复，但是不能保证数据不丢失。
  

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

• 幂等性原理
  幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。 正确一次（ExactlyOnce） = 幂等性 + 至少一次（ack = -1 + 分区副本数 >= 2 + ISR最小副本数量 >= 2）。
  重复数据的判断标准：具有<ID, Partition, SeqNumber>相同主键的消息提交时，Broker只会持久化一条。其中PID是Kafka每次重启都会分配一个新的；Partition 表现分区号；Sequence Number是单调自增的。 所以幂等性只能保证的是在单分区单会话内不重复。
  
  
  
  
• Kafka生产者事务原理
  说明：开启事务，必须开启幂等性。
  总体流程：
  首先Producer向Kafka集群请求PID（实现幂等性所需），成功获取到PID后发送消息到Kafka集群Broker中的Topic中，此时若Producer发起Commit请求，Kafka集群的对应Broker中的Transaction Coordinator（事务和谐器）会持久化该Commit请求到__transaction_state Topic（存储事务信息的特殊主题）中；此时Broker后台发起Commit请求，成功后会将成功信息持久化到__transaction_state中。
  
  
  
  • __transaction_state Topic：默认有50个分区，每个分区负责一部分事务，事务分别是根据Transaction.id的hashcode值%50计算出该事务属于哪一个分区。该分区Leader副本所在的Broker节点即为这个transaction.id对应的Transaction Coordinator节点。
    
  
  
  
  

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相关信息：

• /kafka/brokers/ids [0,1,2] 记录有哪些服务器
  
• /kafka/brokers/topics/first/partitions/0/state{“leader”:1 ,“isr”:[1,0,2] } 记录谁是Leader，有哪些服务器可用
  
• /kafka/controller {“brokerid”:0} 辅助选举Leader
  

总体流程：

• Broker们启动后向ZK注册节点信息并抢占注册Controller；
  
  
  
  • Kafka 集群中有一个 broker 的 Controller 会被选举为 Controller Leader，负责管理集群broker 的上下线，全部topic的分区副天职配和Leader选举等工作。
    
  
  
• 注册成功Controller后开始监听ZK中的brokers/ids节点信息变化
  
• Controller决定Leader选举，原则：在ISR中存活，并在AR中排在前面的优先成为Leader。
  
  
  
  • 比方：AR[1,0,2], ISR[1，0，2]，那么leader就会按照1，0，2的次序轮询。
    
  • Kafka 中副天职为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后Follower找Leader举行同步数据。
    
  
  
• 选举成功后Controller上节点信息到ZK，brokers/ topics/ first /partitions /0 / state
  
• 其他broker节点从ZK同步集群节点信息
  
• 假设Broker1中的Leader挂了，上述Controller会监听到节点变化，会从ZK获取ISR信息，并按照上述原则选举新Leader，最后更新Leader及ISR
  

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

3.2.1 服役退役节点

• 编写topics-to-move.json创建要负载均衡的主题
  
• 天生实验计划：–topics-to-move-json-file topics-to-move.json --broker-list “0,1,2,3” –generate
  
• 实验计划：–reassignment-json-file increase-replication-factor.json –execute
  

3.2.2 Leader 和 Follower 故障处理细节

LEO（Log End Offset）：每个副本的最后一个offset，LEO其实就是最新的offset + 1。
HW（High Watermark）：全部副本中最小的LEO 。

• Follower故障
  
  
  
  o Follower发生故障后会被临时踢出ISR（Leader/Follower存活节点）
  o 这个期间Leader和Follower继承接收数据
  o 待该Follower规复后，Follower会读取当地磁盘记录的前次的HW，并将log文件高于HW的部分截取掉，从HW开始向Leader举行同步。
  o 等该Follower的LEO >= 该Partition的HW，即Follower追上Leader之后，就可以重新加入ISR了。
  
• Leader故障
  
  
  
  o Leader发生故障之后，会从ISR中选出一个新的Leader
  o 为保证多个副本之间的数据一致性，别的的Follower会先将各自的log文件高于HW的部分截掉，然后从新的Leader同步数据。（即：会丢数据）
  

注意：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。
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（压缩）：

• delete日记删除：将逾期数据删除
  
  
  
  • log.cleanup.policy = delete 全部数据启用删除计谋
    （1）基于时间：默认打开。以 segment中全部记录中的最大时间戳作为该文件时间戳。
    （2）基于巨细：默认关闭。超过设置的全部日记总巨细，删除最早的segment。
    log.retention.bytes，默认等于-1，表现无穷大。
    
  
  
• compact日记压缩：对于相同key的差别value值，只保留最后一个版本。
  log.cleanup.policy = compact 全部数据启用压缩计谋
  
  
  
  

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

• Kafka本身是分布式集群，可以接纳分区技能，并行度高
  
• 读数据接纳稀疏索引，可以快速定位要消费的数据
  
• 次序写磁盘：官网有数据表明，同样的磁盘，次序写能到600M/s，而随机写只有100K/s。这与磁盘的机械机构有关，次序写之所以快，是因为其省去了大量磁头寻址的时间。
  
• 页缓存 + 零拷贝技能
  

• PageCache页缓存：Kafka重度依赖底层操纵体系提供的PageCache功能。当上层有写操纵时，操纵体系只是将数据写入PageCache。当读操纵发生时，先从PageCache中查找，假如找不到，再去磁盘中读取。现实上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来利用。
  
• 零拷贝：Kafka的数据加工处理操纵交由Kafka生产者和Kafka消费者处理。KafkaBroker应用层不关心存储的数据，所以就不消走应用层，传输服从高。
  

以读消费数据过程为例：

• 页缓存+非零拷贝技能：
  生产者将数据写到页缓存以及磁盘中，下游消费者发起读数据请求（消费数据），Kafka将数据从页缓存或者磁盘读出来后会交由应用层，然后再发往内核空间中的Socket Cache，最后在通过网卡发送给消费者。
  
• 页缓存+零拷贝技能：
  下游消费者发起读数据请求（消费数据），Kafka将数据从页缓存或者磁盘读出来后直接交由网卡发送到消费者。减少了大量无用的数据传输，提高了读写服从。
  
  
  
  

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消费者。然后各消费者开始消费。

• 每个消费者都会和coordinator保持心跳（默认3s），一旦超时（session.timeout.ms=45s），该消费者会被移除，并触发再均衡；或者消费者处理消息的时间过长（max.poll.interval.ms 5分钟），也会触发再均衡。
  

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而言的。

• 首先对同一个topic中的全部分区按分区号排序，并对消费者按照字母次序排序。
  
• 通过Pattitons数 / consumer数 来决定每个消费者应该消费几个分区。
  
• 假如除不尽，那么前面几个消费者就会多消费一个分区。
  

举例：当前有七个分区，三个消费者，排序后的分区为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会比其他消费者多消费大量分区，容易造成数据倾斜。
再均衡计谋

• 停止掉1号消费者，快速重新发送消息观看结果（45s以内，越快越好）。
  1号消费者挂掉后，消费者组必要按照超时时间45s来判断它是否退出，时间到了45s后，判断它真的退出就会把任务 整体 分配给其他broker实验。
  
• 再次重新发送消息观看结果（45s以后）。
  消费者0已经被踢出消费者组，所以重新按照range方式分配。
  

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

• 停止掉1号消费者，快速重新发送消息观看结果（45s以内，越快越好）。
  1 号消费者的任务会按照RoundRobin的方式，把数据轮询分成 0 、6和 3号分区数据，分别由1号消费者或者2号消费者消费。
  
• 再次重新发送消息观看结果（45s以后）
  消费者0已经被踢出消费者组，所以重新按照RoundRobin方式分配。
  

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

• 停止掉1号消费者，快速重新发送消息观看结果（45s以内，越快越好）。
  1 号消费者的任务会按照粘性规则，尽可能均衡的随机分成 0和 1号分区数据，分别由1号消费者或者2号消费者消费。
  
• 再次重新发送消息观看结果（45s以后）
  消费者0已经被踢出消费者组，消费者0已经被踢出消费者组，所以重新按照粘性方式分配。
  

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提交计谋

• 自动提交offset：为了使我们可以大概专注于本身的业务逻辑，Kafka提供了自动提交offset的功能。虽然自动提交offset十分简单便利，但由于其是基于时间提交的，开发职员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API。
  
• 手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。
  
  
  
  • commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据。
    
  • commitAsync（异步提交）：发送完提交offset请求后，就开始消费下一批数据了。
    
  
  

4.5.2 漏消费和重复消费

• 指定Offset消费：当Kafka中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量 时（比方该数据已被删除），该怎么办？
  
• 指定时间消费：在生产情况中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。 比方要求按照时间消费前一天的数据，怎么处理？
  

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

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

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

4.6.1 消费者事务

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

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

• 假如是Kafka消费能力不敷，则可以思量增加Topic的分区数，而且同时提升消费组的消费者数量，消费者数=分区数。
  
• 假如是下游的数据处理不及时：提高每批次拉取的数量。因为批次拉取数据过少（拉取数据/处理时间 < 生产速率）， 使处理的数据小于生产的数据，也会造成数据积压。
  
  
  
  

5. Kafka-Eagle监控

Kafka-Eagle框架可以监控Kafka集群的整体运行情况，在生产情况中常常利用。
焦点原理和功能：

• 实时监控：Kafka-Eagle 提供了实时监控 Kafka 集群的功能，包括关键性能指标、主题和分区的状态、消费者组的运动等。
  
• 汗青数据分析：除了实时监控，Kafka-Eagle 还支持汗青数据的分析和查询，帮助用户相识 Kafka 集群的趋势和性能变化。
  
• 告警体系：Kafka-Eagle 内置了强大的告警体系，答应用户定义各种告警规则，以便在出现问题时及时采取举措。
  
• 可定制性：Kafka-Eagle 具有高度可定制性，可以根据您的需求举行配置和扩展。您可以自定义监控面板、图表、报告以及数据源。
  
• JMX 监控：为了使 Kafka-Eagle 可以大概网络 Kafka 的性能指标，必要在 Kafka 服务器上开启 JMX 端口，如许 Kafka-Eagle 才能通过 JMX 接口获取 Kafka 的运行时数据。
  
• 多集群支持：Kafka-Eagle 支持管理多个 Kafka 集群，而且可以管理 Kafka 的 Topic（查看、删除、创建等），也可以对消费者状态举行监控。
  
• 用户界面：Kafka-Eagle 提供了一个用户友好的 Web 界面，用于实时显示 Kafka 集群的状态和性能指标，以及配置告警规则。
  
• Kafka SQL：Kafka-Eagle 支持利用 Kafka SQL 实时消费，答应用户通过 SQL 查询 Kafka 中的数据。
  
• 数据保留计谋：Kafka-Eagle 答应用户设置数据的默认保留时间，确保监控数据的及时清理和存储服从。
  
• 告警通知方式：Kafka-Eagle 支持多种告警通知方式，包括邮件、短信、Webhook 等，确保用户可以大概及时收到告警信息。
  

思考：假如不利用Kafka-Eagle如何实现监控Kafka集群功能？
Interceptor拦截器！可以通过在此处设置检查点，同样可以实现监控Kafka集群的功能。
6. Kafka-Kraft模式

6.1 基于ZK与基于KRaft的Kafka

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

• 基于 Zookeeper 的协作模式，使得 Kafka 的集群一致性维护越来越复杂；
  
• 受到 Zookeeper 性能的限制，使得 Kafka 无法支撑更大的集群规模；
  
• 而且 Zookeeper 自身带来的运维复杂性和产品稳定性，也同样将复杂度和风险负担传递到 Kafka 运维职员；
  

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

6.2 新旧模式对比

• 在 Zookeeper （后面简称为 ZK）模式下：
  
  
  
  • 运维摆设：3 个 ZK 节点；2…N 个 Broker 节点，其中一个 Broker 负担 Controller 的角色。除了拉起一套最小生产的 Kafka 集群必要至少 3 + N 的资源外，Kafka 的运维职员要同时把握 ZK 和 Kafka Broker 两套完全差别的体系的运维方式。
    
  • 通讯和谐：ZK 节点之间通过 ZAB 协议举行一致性和谐；Broker 会通过 ZK 来选出一个 Controller 负责全局的和谐，同时也会直接修改 ZK 里的数据；Controller 也会监听和修改 ZK 里的数据，并调用 Broker 来完成集群的和谐。虽然 ZK 之间的一致性由 ZAB 来保障了，但是 ZK 与 Controller 之间和 Controller 与 Broker 之间的一致性是相对比较脆弱的。
    
  
  
• 在 KRaft 模式下：
  
  
  
  • 运维摆设：3 个 Controller 节点；0…N 个 Broker 节点。Kafka 节点可以同时负担 Controller 和 Broker 两个角色，因此一套最小生产集群只必要 3 个节点。在测试情况更可以只以 1 节点模式就可以轻量地拉起一个 Kafka 集群。
    
  • 通讯和谐：Controller 节点底层通过 Raft 协议达成一致，Controller 的内存状态通过 #replay Raft Log 来构建，因此 Controller 之间的内存状态都是一致的；Broker 订阅 KRaft Log 维护和 Controller 一致的内存状态，而且通过事件驱动的方式实验 Partition Reassignment 之类的操纵来实现集群最终一致性和谐。整个集群的状态维护和一致性和谐都是基于 KRaft 中的事件。
    
  
  

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企服之家，中国第一个企服评测及商务社交产业平台。