Kafka问题补充

Posted 2021-06-12 Icedzzz

文章目录

• • 为什么消费组是拉取数据，而不是Kafka自动推送？
  • Kafka于Zookeeper的关系
  • Kafka如何保证数据可靠性？
  • • 复制机制
    • broker配置
  • 如何保证生产者的可靠性？
  • 如何保证消费者的可靠性？
  • 物理存储
  • • 分区分配
    • 文件管理
  • Kafa的事务
  • Kafka为什么快？
  • • 零拷贝sendfile
    • 顺序写
    • Memory Mapped Files(mmap文件映射)
    • kafka中的mmap和零拷贝
  • Kafka为什么吞吐量高？

为什么消费组是拉取数据，而不是Kafka自动推送？

consumer采用pull（拉）模式从broker中读取数据。push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据consumer的消费能力以适当的速率消费消息。
pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据。针对这一点，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间之后再返回，这段时长即为timeout。

Kafka于Zookeeper的关系

• **Kafka 使用 Zookeeper 来维护集群成员的信息。**每个 broker 都有一个唯一标识符，这个标识符可以在配置文件里指定，也可以自动生成。在 broker 启动的时候，它通过创建临时节点把自己的 ID 注册到 Zookeeper。 Kafka 组件订阅 Zookeeper 的 /brokers/ids 路径（broker 在 Zookeeper 上的注册路径），当有 broker 加入集群或退出集群时，这些组件就可以获得通知。
• 如果你要启动另一个具有相同 ID 的 broker，会得到一个错误——新 broker 会试着进行注册，但不会成功，因为 Zookeeper 里已经有一个具有相同 ID 的 broker。
• 在 broker 停机、出现网络分区或长时间垃圾回收停顿时， broker 会从 Zookeeper 上断开连接，此时 broker 在启动时创建的临时节点会自动从 Zookeeper 上移除。监听 broker 列表的Kafka 组件会被告知该 broker 已移除。
• 复制控制器的选举和控制：
  **控制器：**控制器其实就是一个 broker，负责管理集群broker的上下线，所有topic的分区副本分配和leader选举等工作。集群里第一个启动的 broker 通过在Zookeeper 里创建一个临时节点 /controller 让自己成为控制器。其他 broker 在启动时也会尝试创建这个节点，不过它们会收到一个“节点已存在”的异常，然后“意识”到控制器节点已存在，也就是说集群里已经有一个控制器了。其他 broker 在控制器节点上创建Zookeeper watch 对象，这样它们就可以收到这个节点的变更通知。这种方式可以确保集群里一次只有一个控制器存在。
• 如果控制器被关闭或者与Zookeeper断开连接，Zookeeper 上的临时节点就会消失。集群里的其他 broker 通过 watch 对象得到控制器节点消失的通知，它们会尝试让自己成为新的控制器。第一个在 Zookeeper 里成功创建控制器节点的 broker 就会成为新的控制器，其他节点会收到“节点已存在”的异常，然后在新的控制器节点上再次创建 watch 对象。
• 当控制器发现一个 broker 已经离开集群（通过观察相关的 Zookeeper 路径），它就知道，那些失去首领的分区需要一个新首领（这些分区的首领刚好是在这个 broker 上）。控制器遍历这些分区(IST清单)，并确定谁应该成为新首领。
• Kafka 使用 Zookeeper 的临时节点来选举控制器，并在节点加入集群或退出集群时通知控制器。控制器负责在节点加入或离开集群时进行分区首领选举。控制器使用epoch 来避免“脑裂”。“脑裂”是指两个节点同时认为自己是当前的控制器。

如何找到首选首领(分区首领选举)？
从分区的副本(ISR)清单里可以很容易找到首选首领（可以使用 kafka.topics.sh 工具查看副本和分区的详细信息）。清单里的第一个副本一般就是首选首领。

Kafka如何保证数据可靠性？

1. Kafka 可以保证分区消息的顺序。如果使用同一个生产者往同一个分区写入消息，而且消息 B 在消息 A 之后写入， 那么 Kafka 可以保证消息 B 的偏移量比消息 A 的偏移量大，而且消费者会先读取消息 A 再读取消息 B。
2. 只有当消息被写入分区的所有同步副本时，它才被认为是“已提交”的(ack=-1)
3. 只要还有一个副本是活跃的，那么已经提交的消息就不会丢失
4. 消费者只能读取已经提交的消息。
   然而这些均依赖于Kafka的复制机制和分区的多副本架构，这是Kafka可靠性保证的核心。把消息写入多个副本可以使 Kafka 在发生崩溃时仍能保证消息的持久性。

复制机制

Kafka 的主题被分为多个分区，分区是基本的数据块。分区存储在单个磁盘上， Kafka 可以保证分区里的事件是有序的，分区可以在线（可用），也可以离线（不可用）。每个分区可以有多个副本，其中一个副本是首领。所有的事件都直接发送给首领副本，或者直接从首领副本读取事件。其他副本只需要与首领保持同步，并及时复制最新的事件。当首领副本不可用时，其中一个同步副本将成为新首领。
如果跟随者在 10s 内没有请求任何消息，或者虽然在请求消息，但在 10s 内没有请求最新的数据，那么它就会被认为是不同步的。跟随者的正常不活跃时间或在成为不同步副本之前的时间是通过replica.lag.time.max.ms配置的

broker配置

broker 有 3 个配置参数会影响 Kafka 消息存储的可靠性。

1. 复制系数:主 题 级 别 的 配 置 参 数 是 replication.factor， 而 在 broker 级 别 则 可 以 通 过 default.replication.factor 来配置自动创建的主题。如果复制系数为 N，那么在 N-1 个 broker 失效的情况下，仍然能够从主题读取数据或向主题写入数据。所以，更高的复制系数会带来更高的可用性、可靠性和更少的故障。另一方面，复制系数 N 需要至少 N 个 broker，而且会有 N 个数据副本，也就是说它们会占用 N倍的磁盘空间。Kafka 会确保分区的每个副本被放在不同的 broker 上。不过，有时候这样仍然不够安全。如果这些 broker 处于同一个机架上，一旦机架的交换机发生故障，分区就会不可用，这时候把复制系数设为多少都不管用。为了避免机架级别的故障，我们建议把 broker 分布在多个不同的机架上，并使用 broker.rack 参数来为每个broker 配置所在机架的名字。
2. 不完全的首领选举:
   unclean.leader.election 只能在 broker 级别进行配置，它的默认值是 true。当分区首领不可用时，一个同步副本会被选为新首领。如果在选举过程中没有丢失数据，也就是说提交的数据同时存在于所有的同步副本上，那么这个选举就是“完全”的。
   不完全的首领选举即在首领不可用时其他副本都是不同步的情况。当分区其余的跟随副本因broker崩溃、网络问题不可用/不同步时，只有首领作为唯一的同步副本继续接收消息。这个时候，如果首领变为不可用，另外两个副本就再也无法变成同步的了。如果不同步的副本不能被提升为新首领，那么分区在旧首领（最后一个同步副本）恢复之前是不可用的。如果不同步的副本可以被提升为新首领，那么在这个副本变为不同步之后写入旧首领的消息会全部丢失，导致数据不一致
3. 最少同步副本：如上可知，如果“所有副本”只包含一个同步副本，那么在这个副本变为不可用时，数据就会丢失。如果要确保已提交的数据被写入不止一个副本，就需要把最少同步副本数量设置为大一点
   的值。对于一个包含 3 个副本的主题，如果 min.insync.replicas 被设为 2，那么至少要存在两个同步副本才能向分区写入数据。

如何保证生产者的可靠性？

生成者通过Acks值确保发送消息的可靠性：

• acks=0 意味着如果生产者能够通过网络把消息发送出去，那么就认为消息已成功写入Kafka。这样可以得到惊人的吞吐量和带宽利用率，不过如果选择了这种模式，一定会丢失一些消息。
• acks=1 意味着首领在收到消息并把它写入到分区数据文件（不一定同步到磁盘上）时会返回确认或错误响应。在这个模式下，如果发生正常的首领选举，生产者会在选举时收到一个LeaderNotAvailableException 异常，如果生产者能恰当地处理这个错误，它会重试发送消息，最终消息会安全到达新的首领那里。在这个模式下仍然有可能丢失数据，比如消息已经成功写入首领，但在消息被复制到跟随者副本之前首领发生崩溃。
• acks=all 意味着首领在返回确认或错误响应之前，会等待所有同步副本都收到消息。如果和min.insync.replicas 参数结合起来，就可以决定在返回确认前至少有多少个副本能够收到消息。生产者会一直重试直到消息被成功提交。不过这也是最慢的做法，生产者在继续发送其他消息之前需要等待所有副本都收到当前的消息。

如何保证消费者的可靠性？

只有已经被写入所有同步副本的数据，对消费者是可用的，这意味着消费者得到的消息已经具备了一致性。消费者唯一要做的是跟踪哪些消息是已经读取过的，哪些是还没有读取过的。这是在读取消息时不丢失消息的关键。在从分区读取数据时，消费者会获取一批事件，检查这批事件里最大的偏移量，然后从这个偏移量开始读取另外一批事件。这样可以保证消费者总能以正确的顺序获取新数据。
如果消费者提交了偏移量却未能处理完消息，那么就有可能造成消息丢失，这也是消费者丢失消息的主要原因
为了保证消费行为的可靠性，需要注意以下四个参数：

1. group.id ：如果两个消费者具有相同的group.id，并且订阅了同一个主题，那么每个消费者会分到主题分区的一个子集，也就是说它们只能读到所有消息的一个子集（不过群组会读取主题所有的消息）
2. auto.offset.reset：这个参数指定了在没有偏移量可提交时或者请求的偏移量在 broker 上不存在时，消费者会做些什么。这个参数有两种配置。一种是 earliest，如果选择了这种配置，消费者会从分区的开始位置读取数据，不管偏移量是否有效，这样会导致消费者读取大量的重复数据，但可以保证最少的数据丢失。一种是 latest，如果选择了这种配置，消费者会从分区的末尾开始读取数据，这样可以减少重复处理消息，但很有可能会错过一些消息。
3. enable.auto.commit ：这是一个非常重要的配置参数，你可以让消费者基于任务调度自动提交偏移量，也可以在代码里手动提交偏移量。自动提交的一个最大好处是，在实现消费者逻辑时可以少考虑一些问题。如果你在消费者轮询操作里处理所有的数据，那么自动提交可以保证只提交已经处理过的偏移量。自动提交的主要缺点是，无法控制重复处理消息（比如消费者在自动提交偏移量之前停止处理消息），而且如果把消息交给另外一个后台线程去处理，自动提交机制可能会在消息还没有处理完毕就提交偏移量。
4. auto.commit.interval.ms：如果选择了自动提交偏移量，可以通过该参数配置提交的频度，默认值是每 5 秒钟提交一次。一般来说，频繁提交会增加额外的开销，但也会降低重复处理消息的概率。

显式提交偏移量：

• 总是在处理完事件后再提交偏移量
• 提交频度是性能和重复消息数量之间的权衡
• 再均衡
• 仅一次传递
  实现仅一次处理最简单且最常用的办法是把结果写到一个支持唯一键的系统里，在这种情况下，要么消息
  本身包含一个唯一键（通常都是这样），要么使用主题、分区和偏移量的组合来创建唯一键——它们的组合可以唯一标识一个 Kafka 记录。如果你把消息和一个唯一键写入系统，然后碰巧又读到一个相同的消息，只要把原先的键值覆盖掉即可。数据存储引擎会覆盖已经存在的键值对，就像没有出现过重复数据一样。这个模式被叫作幂等性写入。

物理存储

• Kafka中消息是以topic进行分类的，生产者生产消息，消费者消费消息，都是面向topic的。

• topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是producer生产的数据。

• Producer生产的数据会被不断追加到该log文件末端，且每条数据都有自己的offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个offset，以便出错恢复时，从上次的位置继续消费。

• Kafka 的基本存储单元是分区。分区无法在多个 broker 间进行再细分，也无法在同一个broker 的多个磁盘上进行再细分。

• 由于生产者生产的消息会不断追加到log文件末尾，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment对应两个文件——“.index”文件和“.log”文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号。例如，first这个topic有三个分区，则其对应的文件夹为first-0,first-1,first-2。
  
  index和log文件以当前segment的第一条消息的offset命名。下图为index文件和log文件的结构示意图。
  

• “.index”文件存储大量的索引信息，“.log”文件存储大量的数据，索引文件中的元数据指向对应数据文件中message的物理偏移地址。

分区分配

在创建主题时， Kafka 首先会决定如何在 broker 间分配分区。

• 在 broker 间平均地分布分区副本。
• 确保每个分区的每个副本分布在不同的 broker 上。
• 如果为 broker 指定了机架信息，那么尽可能把每个分区的副本分配到不同机架的 broker上。
• 为分区和副本选好合适的 broker 之后，接下来要决定这些分区应该使用哪个目录。我们单独为每个分区分配目录，规则很简单：计算每个目录里的分区数量，新的分区总是被添加到数量最小的那个目录里。

文件管理

• 保留数据是 Kafka 的一个基本特性， Kafka 不会一直保留数据，也不会等到所有消费者都读取了消息之后才删除消息.Kafka 管理员为每个主题配置了数据保留期限，规定数据被删除之前可以保留多长时间，或者清理数据之前可以保留的数据量大小。
• 在一个大文件里查找和删除消息是很费时的，也很容易出错，所以我们把分区分成若干个片段。默认情况下，每个片段包含 1GB 或一周的数据，以较小的那个为准。在 broker往分区写入数据时，如果达到片段上限，就关闭当前文件，并打开一个新文件。
• 当前正在写入数据的片段叫作活跃片段。活动片段永远不会被删除
• 文件格式：我们把 Kafka 的消息和偏移量保存在文件里。保存在磁盘上的数据格式与从生产者发送过来或者发送给消费者的消息格式是一样的。因为使用了相同的消息格式进行磁盘存储和网络传输， Kafka 可以使用零复制技术给消费者发送消息，同时避免了对生产者已经压缩过的消息进行解压和再压缩。
• 零复制技术是指计算机执行操作时，CPU不需要先将数据从某处内存复制到另一个特定区域。
• 如果生产者发送的是压缩过的消息，那么同一个批次的消息会被压缩在一起，被当作“包装消息”进行发送。broker 就会收到一个这样的消息，然后再把它发送给消费者。消费者在解压这个消息之后，会看到整个批次的消息，它们都有自己的时间戳和偏移量。
  - 如果在生产者端使用了压缩功能（极力推荐），那么发送的批次越大，就意味着在网络传输和磁盘存储方面会获得越好的压缩性能。但如果修改了消费者使用的消息格式，那么网络传输和磁盘存储的格式也要随之改变。

Kafa的事务

引用：Kafka的事务

Kafka的事务和数据库里用的ACID事务不是一个东西。Kafka的事务主要用来处理consume-process-produce场景的原子性问题：一个数据处理服务从Kafka的若干源topic取数据，处理后，再发送到另外一些目标topic里。在这个过程中，Kafka事务保证：要么数据被处理了，目标topic的结果被正确写入，源topic的数据被消费掉；要么这个数据还能从源topic里读取到，就像没被处理过一样，不会出现源topic还没consume，目标topic已经produce出去的情况。并且，源和目标topic不止一个时也可以保证这个特性。

生成者事务：多分区原子性写入保证producer发送到多个分区的一批消息要么都成功要么都失败=>所谓的失败是指对事务型consumer不可见；而consumer端读取事务消息主要由consumer端隔离级别体现，它类似于数据库中隔离级别的概念，目前只是简单分为：read_uncommitted和read_committed，其中后者指的是consumer只能读取已成功提交事务的消息（当然也包括非事务型producer生产的消息）

• **消费者事务:**如果想完成Consumer端的精准一次性消费，那么需要kafka消费端将消费过程和提交offset过程做原子绑定。此时我们需要将kafka的offset保存到支持事务的自定义介质中（比如mysql）

Kafka为什么快？

kafka作为MQ也好，作为存储层也好，无非是两个重要功能，一是Producer生产的数据存到broker，二是 Consumer从broker读取数据；我们把它简化成如下两个过程：

• 网络数据持久化到磁盘 (Producer 到 Broker)
• 磁盘文件通过网络发送（Broker 到 Consumer）

零拷贝sendfile

零拷贝并不是不需要拷贝，而是**减少不必要的拷贝次数。**通常是说在IO读写过程中。
实际上，零拷贝是有广义和狭义之分，目前我们通常听到的零拷贝，包括上面这个定义减少不必要的拷贝次数都是广义上的零拷贝。

传统的IO模型通常需要四次copy过程：

1. 第一次：将磁盘文件，读取到操作系统内核缓冲区；
2. **第二次：**将内核缓冲区的数据，copy到application应用程序的buffer；
3. **第三次：**将application应用程序buffer中的数据，copy到socket网络发送缓冲区(属于操作系统内核的缓冲区)；
4. **第四次：**将socket buffer的数据，copy到网卡，由网卡进行网络传输。
   传统方式，读取磁盘文件并进行网络发送，经过的四次数据copy是非常繁琐的。实际IO读写，需要进行IO中断，需要CPU响应中断(带来上下文切换)，尽管后来引入DMA来接管CPU的中断请求，但四次copy是存在“不必要的拷贝”的。
   实际上并不需要第二个和第三个数据副本。应用程序除了缓存数据并将其传输回套接字缓冲区之外什么都不做。相反，数据可以直接从读缓冲区传输到套接字缓冲区。显然，第二次和第三次数据copy 其实在这种场景下没有什么帮助反而带来开销，这也正是零拷贝出现的意义。

**零拷贝 ：是指数据直接在内核完成输入和输出，不需要拷贝到用户空间再写出去，读取磁盘文件后，不需要做其他处理，直接用网络发送出去。**试想，如果读取磁盘的数据需要用程序进一步处理的话，必须要经过第二次和第三次数据copy，让应用程序在内存缓冲区处理。

顺序写

磁盘读取时间主要包括：

• 寻道时间，表示磁头在不同磁道之间移动的时间。

• 旋转延迟，表示在磁道找到时，中轴带动盘面旋转到合适的扇区开头处。

• 传输时间，表示盘面继续转动，实际读取数据的时间。

• 顺序读写，磁盘会预读，预读即在读取的起始地址连续读取多个页面，主要时间花费在了传输时间，而这个时间两种读写可以认为是一样的。

• 随机读写，因为数据没有在一起，将预读浪费掉了。需要多次寻道和旋转延迟。而这个时间可能是传输时间的许多倍。

磁盘顺序读或写的速度400M/s，能够发挥磁盘最大的速度。
随机读写，磁盘速度慢的时候十几到几百K/s。这就看出了差距。
kafka将来自Producer的数据，顺序追加在partition，partition就是一个文件，以此实现顺序写入。
Consumer从broker读取数据时，因为自带了偏移量，接着上次读取的位置继续读，以此实现顺序读。

Memory Mapped Files(mmap文件映射)

参考：http://blog.chinaunix.net/uid-26669601-id-5599887.html
mmap文件映射指的是：主要是只 硬盘上文件 的位置与进程 逻辑地址空间 中一块大小相同的区域之间的一一对应。这种对应关系纯属是逻辑上的概念，物理上是不存在的，原因是进程的逻辑地址空间本身就是不存在的。
虚拟映射只支持文件；

• mmap文件的作用是：将磁盘文件映射到内存, 用户通过修改内存就能修改磁盘文件。
• 它的工作原理是直接利用操作系统的Page来实现文件到物理内存的直接映射。完成映射之后你对物理内存的操作会被同步到硬盘上
• 在进程 的非堆内存开辟一块内存空间，和OS内核空间的一块内存进行映射，
• kafka数据写入、是写入这块内存空间，但实际这块内存和OS内核内存有映射，也就是相当于写在内核内存空间了，且这块内核空间、内核直接能够访问到，直接落入磁盘。
• 使用这种方式可以获取很大的I/O提升，省去了用户空间到内核空间复制的开销。
• mmap也有一个很明显的缺陷——不可靠，写到mmap中的数据并没有被真正的写到硬盘，操作系统会在程序主动调用flush的时候才把数据真正的写到硬盘。

kafka中的mmap和零拷贝

1. 写入Broker
数据落盘通常都是非实时的，kafka生产者数据持久化也是如此。Kafka的数据并不是实时的写入硬盘，它充分利用了现代操作系统分页存储来利用内存提高I/O效率。

对于kafka来说，Producer生产的数据存到broker，这个过程读取到socket buffer的网络数据，**其实可以直接在OS内核缓冲区，完成落盘。并没有必要将socket buffer的网络数据，读取到应用进程缓冲区；**在这里应用进程缓冲区其实就是broker，broker收到生产者的数据，就是为了持久化。
在此特殊场景下：接收来自socket buffer的网络数据，应用进程不需要中间处理、直接进行持久化时。——可以使用mmap内存文件映射。

2. Broker到Consumer

传统方式实现：先读取磁盘、再用socket发送，实际也是进过四次copy。
而 Linux 2.4+ 内核通过 sendfile 系统调用，提供了零拷贝。**磁盘数据通过 DMA 拷贝到内核态 Buffer 后，直接通过 DMA 拷贝到 NIC Buffer(socket buffer)，无需 CPU 拷贝。**这也是零拷贝这一说法的来源。除了减少数据拷贝外，因为整个读文件 - 网络发送由一个 sendfile 调用完成，整个过程只有两次上下文切换，因此大大提高了性能。零拷贝过程如下图所示。

总的来说Kafka快的原因：

1. partition顺序读写，充分利用磁盘特性，这是基础；
2. Producer生产的数据持久化到broker，采用mmap文件映射，实现顺序的快速写入；
3. Customer从broker读取数据，采用sendfile，将磁盘文件读到OS内核缓冲区后，直接转到socket buffer进行网络发送。

Kafka为什么吞吐量高？

除了上述磁盘顺序写和零拷贝，还有个重要的理由：数据以分区形式存放
Kafka的message是按topic分类存储的，topic中的数据又是按照一个一个的partition即分区存储到不同broker节点。每个partition对应了操作系统上的一个文件夹，partition实际上又是按照segment分段存储的。这也非常符合分布式系统分区分桶的设计思想。

通过这种分区分段的设计，Kafka的message消息实际上是分布式存储在一个一个小的segment中的，每次文件操作也是直接操作的segment。为了进一步的查询优化，Kafka又默认为分段后的数据文件建立了索引文件，就是文件系统上的.index文件。这种分区分段+索引的设计，不仅提升了数据读取的效率，同时也提高了数据操作的并行度。

引用：
Kafka权威指南
https://zhuanlan.zhihu.com/p/88789697
尚硅谷Kafka文档