Kafka基础知识

r囧r小猫 2024-04-01 14:32 31阅读 0赞

参考博客：https://blog.csdn.net/nandao158/article/details/124001362  
**kafka producer如何优化打入速度?**  
增加线程  
提高 batch.size  
增加更多 producer 实例  
增加 partition 数  
设置 acks=-1 时，如果延迟增大：可以增大 num.replica.fetchers（follower 同步数据的线程数）来调解；  
**Kafka 分区数可以增加或减少吗？说说你自己的理解？**  
只能增加不能减少  
如果减少分区会报The number of partitions for a topic can only be increased.  
减少的分区其数据放到哪里去？是删除，还是保留？删除的话，那么这些没消费的消息就丢了。如果保留这些消息如何放到其他分区里面？追加到其他分区后面的话那么就破坏了 Kafka 单个分区的有序性。  
**kafka的分区为什么没有设计读写分离?**  
主写从读也就是读写分离，可以均摊一定的负载，却不能做到完全的负载均衡。  
比如对于数据写压力很大而读压力很小的情况，从节点只能分摊很少的负载压力，而绝大多数压力还是在主节点上。  
而在 Kafka 中却可以达到很大程度上的负载均衡，而且这种均衡是在主写主读的架构上实现的。  
Kafka 的生产消费模型，如下图所示。  
![在这里插入图片描述][326c108776b644f7bad097626f64ffb8.jpeg_pic_center]

**在Kafka集群中有3个分区，每个分区有3个副本，正好均匀地分布在 3个 broker 上。**  
灰色阴影的代表 leader 副本，非灰色阴影的代表 follower 副本，虚线表示 follower 副本从 leader 副本上拉取消息。

当生产者写入消息的时候，都写入 leader 副本，对于图中的情形，每个 broker 都有消息从生产者流入。  
当消费者读取消息的时候，也是从 leader 副本中读取的，对于图中的情形，每个 broker 都有消息流出到消费者。  
我们很明显地可以看出，每个 broker 上的读写负载都是一样的，这就说明 Kafka 可以通过主写主读实现主写从读实现不了的负载均衡。  
**1.可以简化代码的实现逻辑，减少出错的可能;  
2.将负载粒度细化均摊，与主写从读相比，不仅负载效能更好，而且对用户可控;  
3.没有延时的影响;  
4.在副本稳定的情况下，不会出现数据不一致的情况。**

kafka分区的目的  
分区对于 Kafka 集群：实现**负载均衡**。  
分区对于消费者来说，可以**提高并发度，提高吞吐量**。

kafka和传统消息系统的差异  
• 首先，kafka是个现代分布式系统，以集群的方式运行，可以自由伸缩  
• 其次，kafka可以按照要求存储数据，保存多久都可以  
• 第三，流式处理将数据处理的层次提示到了新高度，消息系统只会传递数据。kafka的流式处理能力可以让我们用很少的代码就能动态的处理派生流和数据集。所以，kafka不仅仅是个消息中间件  
kafka不仅仅是个消息中间件，同时它是个流平台，这个平台上可以发布和订阅数据流（kafka的流，有一个单独的包stream的处理），并把它们保存起来，进行处理，这个就是kafka作者的设计理念

**Kafka监控工具**  
Kafka manager  
Kafka monitor  
Jmx tool

**Kafka简介**  
**1、简介**  
Kafka 是用于日志处理的分布式消息队列，同时支持离线和在线日志处理。kafka 对消息保存时根据Topic进行归类，发送消息者成为Producer,消息接受者成为Consumer,此外kafka 集群由多个kafka 实例组成，每个实例(server)称为broker。无论是kafka集群，还是producer和consumer 都依赖于zookeeper 来保证系统可用性，为集群保存一些meta 信息（元信息）。  
**·** Kafka维护消息类别的东西是主题（topic）.  
**·** 我们称发布消息到Kafka主题的进程叫生产者（producer）.  
**·** 我们称订阅主题、获取消息的进程叫消费者（consumer）.  
**·** Kafka是由多个服务器组成的机器，每个服务器称作代理（broker）  
![在这里插入图片描述][9bfc1b2d757d4b248c58ff958c5fc80b.jpeg_pic_center]

客户端与服务器之间的通信通过TCP协议  
Kafka的整体架构如图所示。因为Kafka内在就是分布式的，一个Kafka集群通常包括多个代理。为了均衡负载，将话题分成多个分区，每个代理存储一或多个分区。多个生产者和消费者能够同时生产和获取消息。  
![在这里插入图片描述][4b554cfa723441feb4c6bc0912034b71.jpeg_pic_center]

**2、主题和分区（分区本质上就是一个分区文件目录）**  
Kafka里的消息用主题进行分类，主题下有可以被分为若干个分区。分区本质上是个提交日志，有新消息，这个消息就会以追加的方式写入分区，然后用先入先出的顺序读取。

但是因为主题会有多个分区，所以在整个主题的范围内，是无法保证消息的顺序的，单个分区则可以保证。分区中的1、2、3表示的消息放置的下标位置，就是下文所谓的偏移量（位移）。  
  Kafka通过分区来实现数据冗余和伸缩性，因为分区可以分布在不同的服务器上，那就是说一个主题可以跨越多个服务器。  
  前面我们说Kafka可以看成一个流平台，很多时候，我们会把一个主题的数据看成一个流，不管有多少个分区。  
**1.Kafka将主题下的分区分配给消费组里的消费者，每个分区被一个消费者消费**  
**2.消费者的数量不能超过分区数**  
**3.Kafka只能保证分区内的消息是有序的**  
**4.如果你想要消息是全局有序的，你可以设置主题只有一个分区，同时这意味着只能有一个消费者**

**kafka使用场景**  
1、 消息传输  
2、 网站行为日志追踪  
3、 审计数据收集  
4、 日志收集

**消息系统**： Kafka 和传统的消息系统一样都具备系统解耦、冗余存储、流量削峰、缓冲、异步通信、扩展性、可恢复性等功能。与此同时，Kafka 还提供了大多数消息系统难以实现的消息顺序性保障及回溯消费的功能。  
**存储系统**： Kafka 把消息持久化到磁盘，相比于其他基于内存存储的系统而言，有效地降低了数据丢失的风险。也正是得益于 Kafka 的消息持久化功能和多副本机制，我们可以把 Kafka 作为长期的数据存储系统来使用，只需要把对应的数据保留策略设置为“永久”或启用主题的日志压缩功能即可。  
**流式处理平台**： Kafka 不仅为每个流行的流式处理框架提供了可靠的数据来源，还提供了一个完整的流式处理类库，比如窗口、连接、变换和聚合等各类操作。

**Kafka是如何实现高吞吐率的？**  
**1）三层网络架构设计**  
Kafka的网络通信模型，主要采用了1（1个Acceptor线程）+ N（N个Processor线程）+ M（M个业务处理线程）。  
可以看出Kafka的Broker NIO异步并发处理消息，实现了IO线程异步并发处理消息的机制，大大提升了数据的吞吐量。  
**2）顺序读写**  
**3）零拷贝**  
非零拷贝的情况要经过用户缓冲区user space  
![在这里插入图片描述][6eb2c84e7152406291fe5620bde9bb92.jpeg_pic_center]

在Linux kernel2.2之后出现了一种叫做“零拷贝（zero-copy）”系统调用机制，就是跳过“用户缓冲区”的拷贝，建立一个磁盘空间和内存的直接映射，数据不再复制到“用户态缓冲区”。  
系统上下文切换减少为2次，可以提升一倍的性能；

**4)文件分段**  
Kafka的队列Topic被分为多个Partition，每个Partition又分为多段segment，所以一个队列中的消息实际上是保存在N个多个片段文件中；通过分段的方式，每次文件操作都是对一个小文件的操作，非常轻便，同时也增加了并行处理的能力；  
**5)批量发送**  
Kafka允许进行批量发送消息，先将消息缓存在内存中，然后一次请求批量发送出去比如可以指定缓存的消息达到某个量就发出去，或者缓存了固定的时间后就发送出去，如100条消息就发送一次，或者每5秒发送一次。这种策略将大大减少服务端的I/O次数  
**6)数据压缩**  
Kafka还支持对消息集合进行压缩，Producer可以通过GZIP或Snappy格式对消息集合进行压缩，压缩的好处就是减少传输的数据量，减轻对网络传输的压力。  
Producer压缩之后，在Consumer需要进行解压，虽然增加了CPU的工作，但是在大数据处理上瓶颈在网络，而不是在CPU，这个成本比较值得。

**Kafka为什么读写效率高**  
1.分区partition并行处理  
2.顺序写磁盘，充分利用磁盘特性  
3.利用了现代操作系统分页存储 Page Cache 来利用内存提高 I/O 效率  
4.采用了零拷贝技术。Producer 生产的数据持久化到 broker，采用 mmap 文件映射，实现顺序的快速写入。Customer 从 broker 读取数据，采用 sendfile，将磁盘文件读到 OS 内核缓冲区后，转到 NIO buffer进行网络发送，减少 CPU 消耗。

**kafka文件高效存储设计特点**  
**文件分段**： Kafka把topic中一个partition大文件分成多个小文件段（默认1G），通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用。  
**跳表设计**： 通过索引信息可以快速定位message和确定response的最大大小。  
**索引元数据存内存**：通过index元数据全部映射到memory，可以避免IO磁盘操作。  
**稀疏索引**： 通过索引文件稀疏存储，可以大幅降低index文件元数据占用空间大小。

**Kafka中的分区器、序列化器、拦截器是否了解？它们之间的处理顺序是什么？**  
整个kafka生产者客户端由两条线程协调运行。  
这两条线程分别为**主线程**和**sender线程（发送线程）**  
主线程的作用就是：由KafkaProducer创建消息，然后通过可能的拦截器-》序列化器-》分区器的作用之后缓存到消息累加器  
send线程的作用就是：负责将消息累加器中的消息发送到kafka中。  
生产者拦截器可以用来在消息发送前做一些准备工作，比如按照某个规则过滤不符合要求的消息，修改消息的内容等，也可以用来在发送回调逻辑前做一些定制化的需求。  
生产者需要用序列化器将对象转换成字节数组才能通过网络发送给kafka.当然消费者需要用对应的反序列化器将kafka的字节数组转换为相应的对象。  
经过拦截器，序列化器，之后就会需要确定消息要发往的分区，这里会用到分区器，有默认分区器和自定义分区器。

**kafka中生产数据的时候，如何设置参数保证消息写入的正确性？**  
**1.Topic 分区副本**  
Kafka 的分区多副本架构是 Kafka 可靠性保证的核心，把消息写入多个副本可以使 Kafka 在发生崩溃时仍能保证消息的持久性。  
**2.Producer往Broker发送消息**  
Kafka 在 Producer 里面提供了消息确认机制。也就是说我们可以通过配置来决定消息发送到对应分区的几个副本才算消息发送成功。通过 request.required.acks 参数设置，参数支持以下三种值：  
acks = 0：意味着如果生产者能够通过网络把消息发送出去，那么就认为消息已成功写入 Kafka 。  
acks = 1：意味若 Leader 在收到消息并把它写入到分区数据文件（不一定同步到磁盘上）时会返回确认或错误响应。  
acks = all（这个和 request.required.acks = -1 含义一样）：意味着 Leader 在返回确认或错误响应之前，会等待所有同步副本都收到消息。  
根据实际的应用场景，我们设置不同的 acks，以此保证数据的可靠性。  
**3.Producer 发送消息还可以选择同步（默认，通过 producer.type=sync 配置） 或者异步（producer.type=async）模式。**  
如果设置成异步，虽然会极大的提高消息发送的性能，但是这样会增加丢失数据的风险。如果需要确保消息的可靠性，必须将 producer.type 设置为 sync。  
**4.Leader 选举**  
每个分区的 leader 会维护一个 ISR 列表，ISR 列表里面就是 follower 副本的 Broker 编号，只有跟得上 Leader 的 follower 副本才能加入到 ISR 里面，这个是通过 replica.lag.time.max.ms 参数配置的（延迟时间）。只有 ISR 里的成员才有被选为 leader 的可能。  
当 Leader 挂掉了，而且 unclean.leader.election.enable=false 的情况下，Kafka 会从 ISR 列表中选择第一个 follower 作为新的 Leader，因为这个分区拥有最新的已经 committed 的消息。通过这个可以保证已经 committed 的消息的数据可靠性。  
**综上所述，为了保证数据的可靠性，我们最少需要配置一下几个参数：  
producer 级别：acks=all（或者 request.required.acks=-1），同时发生模式为同步 producer.type=sync  
topic 级别：设置 replication.factor>=3，并且 min.insync.replicas>=2  
broker 级别：关闭不完全的 Leader 选举，即 unclean.leader.election.enable=false**

**kafka重复消费的原因及解决方法**  
**重复消费的原因**  
原因1：消费者宕机、重启或者被强行kill进程，导致消费者消费的offset没有提交。  
原因2：设置enable.auto.commit为true，如果在关闭消费者进程之前，取消了消费者的订阅，则有可能部分offset没提交，下次重启会重复消费。  
原因3：消费后的数据，当offset还没有提交时，Partition就断开连接。比如，通常会遇到消费的数据，处理很耗时，导致超过了Kafka的session timeout.ms时间，那么就会触发reblance重平衡，此时可能存在消费者offset没提交，会导致重平衡后重复消费。

**重复消费的解决方法**  
1、提高消费者的处理速度。例如：对消息处理中比较耗时的步骤可通过1、多线程异步处理等。在缩短单条消息消费的同时，根据实际场景2、可将max.poll.interval.ms值设置大一点，避免不必要的Rebalance。可根据实际消息速率适当调小max.poll.records的值。

2、引入消息去重机制。例如：生成消息时，在消息中加入唯一标识符如消息id等。在消费端，可以保存最近的max.poll.records条消息id到redis或mysql表中，这样在消费消息时先通过查询去重后，再进行消息的处理。

max.poll.interval.ms：消费者组中的一员在拉取消息时如果超过了设置的最大拉取时间，则会认为消费者消费消息失败，kafka会重新进行重新负载均衡，以便把消息分配给另一个消费组成员  
max.poll.records：每次最大消费消息数量

**Kafka数据积压**  
参考博客：https://www.cnblogs.com/singleYao/p/16419476.html  
处理方法：  
**1.** 实时/消费任务挂掉导致的消费滞后  
a. 任务重新启动后直接消费最新的消息，对于"滞后"的历史数据采用离线程序进行"补漏"。  
b. 任务启动从上次提交offset处开始消费处理  
如果积压的数据量很大，需要增加任务的处理能力，比如增加资源，让任务能尽可能的快速消费处理，并赶上消费最新的消息  
**2.** Kafka分区少了  
如果数据量很大，合理的增加Kafka分区数是关键。如果利用的是Spark流和Kafka direct approach方式，也可以对Kafka RDD进行repartition重分区，增加并行度处理。  
**3.** 由于Kafka消息key设置的不合理，导致分区数据不均衡  
可以在Kafka producer处，给key加随机后缀，使其均衡。

**比较RabbitMQ与Apache Kafka**  
![在这里插入图片描述][ab77b68d149c48238733b7a6dad99659.png_pic_center]

**如何保证kafka数据不丢？**  
**Producer端：**  
1)unclean.leader.election.enable=false （在server.properties文件中配置）  
2)min.insync.replicas=总副本数-1 ，即最小同步副本个数=总副本个数-1（在server.properties文件中配置）  
3)acks=-1/all，即当所有isr副本数全部收到消费之后再提交ack （在producer端代码里配置）  
4)增大副本数。  
5）对于一些由于网络故障等造成发送失败的可重试异常，可以通过设置重试次数（retries）来增加可靠性。  
6）kafka producer.send(message,Callback) 可以通过回调函数，来处理发送失败的数据。  
**Consumer端：**  
消费端尽量保证手动处理偏移量。保证数据能成功消费，不会造成数据丢失的情况。

**说出三个Kafka 与传统消息系统之间的关键区别**  
(1)Kafka 持久化日志，\*\*这些日志可以被重复读取和无限期保留  
(2)Kafka 是一个分布式系统：它以集群的方式运行，可以灵活伸缩，在内部通过，复制数据提升容错能力和高可用性  
(3)Kafka 支持实时的流式处理

**Kafka Rebalance详解**  
**1.rebalance概览**  
它规定了一个 consumer group 是如何达成一致来分配订阅 topic 的所有分区的。比方说Consumer Group A 有3个consumer 实例，它要消费一个拥有6个分区的topic，每个consumer消费2个分区，这就是rebalance。  
rebalance是相对于consumer group 而言，每个consumer group会从kafka broker中选出一个作为组协调者（group coordinator）。coordinator负责对整个consumer group的状态进行管理，当有触发rebalance的条件发生时，促使生成新的分区分配方案。  
**2.rebalance触发条件**  
rebalance触发的条件有三个：

1、consumer group成员发生变更，比方说有新的consumer实例加入，或者有consumer实例离开组，或者有consumer实例发生奔溃。  
2、consumer group订阅的topic数发生变更，这种情况主要发生在基于正则表达式订阅topic情况，当有新匹配的topic创建时则会触发rebalance。  
3、consumer group 订阅的topic分区数发生变更。  
其实无论哪种触发条件，我们可以发现根本原因还是因为topic 中partition或者consumer实例发生了变更。

**3.rebalance分区分配策略**  
分区分配策略决定了将topic中partition分配给consumer group中consumer实例的方式。  
可以通过消费者客户端参数partition.assignment.strategy来设置消费者与主题之间的分区分配策略  
kafka新版本提供了三种rebalance分区分配策略：  
• range  
• round-robin  
• sticky  
range（范围）分配策略的原理是按消费者总数和分区总数进行整除运算来获得一个跨度，然后将分区按照跨度进行平均分配，以保证分区尽可能平均的分配给所有的消费者。

假设 n = 分区数/消费者数量，m= 分区数%消费者数量，那么前m个消费者每个分配n+1个分区，后面的（消费者数量-m）个消费者每个分配n个分区。

round-robin（循环）分配策略是将消费者组内所有主题的分区按照字典序排序，然后通过轮询的方式逐个将分区一次分配给每个消费者。

Sticky（粘稠）分配策略是从0.11.x版本开始引入的分配策略，它主要有两个目的：

（1）分区的分配要尽可能均匀。

（2）分区的分配尽可能与上次分配的保持相同。

当两者发生冲突时，第一个目标优于第二个目标。sticky具体实现要比上面两种要复杂的多。

我们以一个具体的例子来说明。

假设消费者组内有3个消费者（C0、C1、C2），他们都订阅了4个主题（t0、t1、t2、t3）,并且每个主题有两个分区。也就是说，整个消费者组订阅了t0p0、t0p1、t1p0、t1p1、t2p0、t2p1、t3p0、t3p1 8个分区。最终的分配结果为  
消费者C0：t0p0、t1p1、t3p0  
消费者C1：t0p1、t2p0、t3p1  
消费者C2：t1p0、t2p1

这看上去似乎与round robin分配策略相同，事实上并不是这样。假设此时C1脱离了消费者组，那么消费者组就会执行rebalance,进而消费分区会重新分配。如果采用round robin策略，那么此时的分配结果如下  
消费者C0：t0p0、t1p0、t2p0、t3p0  
消费者C2：t0p1、t1p1、t2p1、t3p1

如果采用sticky分配策略，那么分配结果为  
消费者C0：t0p0、t1p1、t3p0、t2p0  
消费者C2：t1p0、t2p1、t0p1、t3p1

可以看到sticky分配结果中保留了上一次分配中对消费者C0和C2的所有的分配结果，并将原来的消费者C1的负担分配给了剩余的两个消费者C0和C2，最终C0和C2的分配还保持了平衡。  
**4.rebalance generation**  
rebalance generation用于标识某次rebalance。它是一个整数，从0开始。它主要是为了保护consumer group的，比如上一届的consumer由于某些原因延迟提交了offset，但rebalance之后该group产生了新的一届成员，而这次延迟的offset提交的是旧的generation信息，因此会被consumer group拒绝。  
**5.rebalance协议**  
rebalance本质上是一组协议，consumer group和coordinator使用这组协议共同完成consumer group的rebalance。kafka新版本提供了下面5个协议来处理Rebalance

JoinGroup请求：consumer请求加入组。  
SyncGroup请求：group leader把分配方案同步更新到组内所有成员中。  
Heartbeat请求：consumer定期向coordinator汇报心跳表明自己依然存活。  
LeaveGroup请求：consumer主动通知coordinator自己将要离开consumer group。  
DescribeGroup请求：查看组的所有的所有信息，包括成员信息、协议信息、分配方案、以及订阅信息等。该请求主要供管理员使用，coordinator不使用该请求实现rebalance。  
rebalance过程中， coordinator 要处理 consumer 发过来的 JoinGroup 和SyncGroup 请求。当consumer 主动离组时会发送 LeaveGroup 请求给 coordinator

在成功rebalance之后，组内所有 consumer 都需要定期地向 coordinator 发送 Heartbeat 请求，而每个 consumer 也是根据 Heartbeat 请求的响应中是否包含 REBALANCE\_IN\_PROGRESS判断当前 group 开启了新一轮 rebalance。  
**6.rebalance流程**  
consumer group在执行rebalance之前必须首先确认coordinator在哪个broker上。并创建与该broker通信的socket连接。确定 coordinator 的算法与确定 offset 被提交到consumer offsets 目标分区的算法是相同的 算法如下：

计算 Math.abs(groupID .hash Code) % offsets. topic.num. partitions 参数值（默认是 50) ，假设是 10  
寻找\_\_consumer\_offsets 分区 10 leader 副本所在的 broker ，该 broker 即为这group的coordinator  
成功连接 coordinator 之后便可以执行 rebalance 操作, 目前 rebalance 主要分为两步：加入组和同步更新分配方案

\*\*加入组：\*\*这一步中组内所有consumer（即group.id相同的所有comsumer实例）向coordinator发送JoinGroup请求。当收集全JoinGroup请求，coordinator从中选择一个consumer 作为group的leader，并把所有成员信息以及他们订阅的topic信息发送给leader。需要注意的是leader是consumer group中的一个consumer实例，而coordinator为集群的一个broker。是leader而不是coordinator负责为整个consumer group成员制定分配方案。

\*\*同步更新分配方案：\*\*这一步leader开始制定分配方案，即根据前面提到的分配策略，决定哪个consumer消费哪些topic的哪些分区，一旦分配完成，leader会将分配方案封装进SyncGroup请求发送给coordinator。值得注意的是，组内所有成员都会发送SyncGroup请求，不过只有leader发送的请求中包含分配方案。coordinator收到分配方案后把属于每个consumer的分配方案单独抽取出来做作为SyncGroup请求的response返回给各个consumer。  
下图分别描述了加入组和同步分配方案的流程。  
![在这里插入图片描述][4eabeeedb8db487db8a9bc80adf502f5.jpeg_pic_center]

[326c108776b644f7bad097626f64ffb8.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/cfacc9b11bf64dca963045f5eaf8c44c.jpeg
[9bfc1b2d757d4b248c58ff958c5fc80b.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/f5b86045ec384c9aa0db7a067843dbb4.jpeg
[4b554cfa723441feb4c6bc0912034b71.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/d79aa1f83fcc4c6e98f2e8ee7ebe014d.jpeg
[6eb2c84e7152406291fe5620bde9bb92.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/a606345c6bef495b975a883de55692c3.jpeg
[ab77b68d149c48238733b7a6dad99659.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/bbc824f16e7f4d85b96fa06d0ac50d28.png
[4eabeeedb8db487db8a9bc80adf502f5.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/4f5f066fb63943499e6675a5b339dd51.jpeg