Kafka 深入(3)：生产者，消费者，分区策略，zookeeper的作用，高效读写的原因

Kafka 深入(3)：生产者，消费者，分区策略，zookeeper的作用，高效读写的原因 Kafka 工作流程及文件存储机制

Topic 与 partition

物理层面：

• topic 是逻辑上的概念，而 partition 是物理上的概念
• 每个 partition 对应于一个 log 文件，该 log 文件中存储的就是 producer 生产的数据。
• Producer 生产的数据会被不断追加到该 log 文件末端，且每条数据都有自己的 offset。
• 消费者组中的每个消费者，都会实时记录自己消费到了哪个 offset，以便出错恢复时，从上次的位置继续消费。

逻辑层面：

• 由于生产者生产的消息会不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采取了分片和索引机制，将每个 partition 分为多个 segment。
• 每个 segment 对应两个文件——“.index” 文件和 “.log” 文件。
• .index 文件存储大量的索引信息，.log文件 存储大量的数据，索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。

Kafka 生产者 分区的策略

• 可扩展性与稳定性：方便在集群中扩展，每个 Partition 可以通过调整以适应它所在的机器，而一个 topic 又可以有多个 Partition 组成，因此整个集群就可以适应任意大小的数据了；
• 并发性： 可以提高并发，以 Partition 为单位进行读写。

partition 的取值原则

• 指明 partition 的情况下，直接将指明的值直接作为 partiton 值；
• 没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值；
• 既没有 partition 值又没有 key 值的情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition 值，也就是常说的 round-robin 算法

数据可靠性的保证

• 为保证 producer 发送的数据，能可靠的发送到指定的 topic，topic 的每个 partition 收到 producer 发送的数据后，都需要 向 producer 发送 ack（acknowledgement 确认收到）
• 如果 producer 收到 ack，就会进行下一轮的发送，否则重新发送数据。

副本的同步策略

Kafka 选择了第二种方案，原因如下：

• 同样为了容忍 n 台节点的故障，第一种方案需要 2n+1 个副本，而第二种方案只需要 n+1 个副本，而 Kafka 的每个分区都有大量的数据，第一种方案会造成大量数据的冗余。
• 虽然第二种方案的网络延迟会比较高，但网络延迟对 Kafka 的影响较小

AR、ISR、OSR

• 分区中的所有副本统称为AR（Assigned Replicas）。
• 所有与 leader 副本保持一定程度同步的副本（包括leader副本在内）组成 ISR(In-Sync Replicas)，ISR集合是AR集合中的一个子集。
• 与 leader 副本同步滞后过多的副本（不包括 leader 副本）组成 OSR（Out-of-Sync Replicas）

全同步可能出现的问题的解决方案：ISR

设想以下情景：leader 收到数据，所有 follower 都开始同步数据，但有一个 follower，因为某种故障，迟迟不能与 leader 进行同步，那 leader 就要一直等下去， 直到它完成同步，才能发送 ack。这个问题怎么解决呢？

• Leader 维护了一个动态的 in-sync replica set (ISR)，意为和 leader 保持同步的 follower 集合。
• 当 ISR 中的 follower 完成数据的同步之后，leader 就会给 follower 发送 ack。
• 如果 follower 长时间未向 leader 同 步 数 据 ， 则 该 follower 将 被 踢 出 ISR ， 该时间阈值由 replica.lag.time.max.ms 参数设定。Leader 发生故障之后，就会从 ISR 中选举新的 leader.
• 简单的可以理解为维护了一个动态的可用的从机表

ACK的应答级别

对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等 ISR 中的 follower 全部接收成功。
Kafka 为用户提供了三种可靠性级别，用户根据对可靠性和延迟的要求进行权衡：

• 0：producer 不等待 broker 的 ACK，这一 *** 作提供了一个最低的延迟，broker 一接收到还没有写入磁盘就已经返回，当 broker 故障时有可能丢失数据；
• 1：producer 等待 broker 的 ACK，partition 的 leader 落盘成功后返回 ack，如果在 follower 同步成功之前 leader 故障，那么将会丢失数据.
• -1（all）：producer 等待 broker 的 ack，partition 的 leader 和 follower 全部落盘成功后才返回 ack。但是如果在 follower 同步完成后，broker 发送 ack 之前，leader 发生故障，那么会造成数据重复.

副样本的一致性保证

Log 文件主要使用 HW 和 LEO 来完成数据的更新与同步。

• LEO：指的是每个副本最大的 offset；
• HW：指的是消费者能见到的最大的 offset，ISR 队列中最小的 LEO
  

通过上面两个变量可以很好的处理下面两种故障：

• follower 故障

  • follower 发生故障后会被临时踢出 ISR，待该 follower 恢复后，follower 会读取本地磁盘记录的上次的 HW，并将 log 文件高于HW 的部分截取掉，从 HW 开始向 leader 进行同步。
  • 等该 follower 的 LEO 大于等于该 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了
  • 简单的说，故障后，从故障的时间戳开始同步leader数据直到赶上当前的读文件为止。

• leader 故障

  • leader 发生故障之后，会从 ISR 中选出一个新的 leader
  • 之后，为保证多个副本之间的数据一致性，其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 leader同步数据

注意： 这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。

Exactly-once 机制

Kafka在0.11.0.0之前的版本中只支持 At Least once 和 At Most once 语义，尚不支持Exactly Once语义。

• At Least once 就是设置 ACK 的应答模式为-1。即可以保证数据的可靠性传输但是可能造成数据的重复传输。
• At Most once 就是设置ACK应答模式为-1，每个数据直接接受一次，但是无法保证数据的可靠传输。
• Exactly-once：对于一些非常重要的信息，比如说交易数据，下游数据消费者要求数据既不重复也不丢失。实现 Exactly-once 的核心思想就是利用接口的幂等性（在编程中，一个幂等 *** 作的特点是其任意多次执行所产生的影响均与一次执行的影响相同）。

实现过程：

• 为了实现 Producer 的幂等语义，Kafka 引入了 Producer ID（即PID）和 Sequence Number。

  • 每个新的 Producer 在初始化的时候会被分配一个唯一的 PID，该PID对用户完全透明而不会暴露给用户。
  • 每个producer发送的每一条信息 都会对应一个从 0 开始递增的 Sequence Number

• 类似地，Broker 端也会为每个 维护一个序号，并且每次 Commit 一条消息时将其对应序号递增。

  • 对于接收的每条消息，如果其序号比 Broker 维护的序号（即最后一次Commit的消息的序号）大一，则 Broker 会接受它，否则将其丢弃：
  • 如果消息序号比 Broker 维护的序号大一以上，说明中间有数据尚未写入，也即乱序，此时Broker拒绝该消息，Producer抛出 InvalidSequenceNumber。
  • 如果消息序号小于等于Broker维护的序号，说明该消息已被保存，即为重复消息，Broker直接丢弃该消息，Producer抛出 DuplicateSequenceNumber

• 可以简单的理解为，给生产者本身和其所生产的信息增加版本号，进而控制顺序和不可重复性

Exactly-once 能够解决的问题

• Broker保存消息后，发送ACK前宕机，Producer认为消息未发送成功并重试，造成数据重复。
• 前一条消息发送失败，后一条消息发送成功，前一条消息重试后成功，造成数据乱序。

事务性的实现

为了实现事务，应用程序必须提供一个稳定的（重启后不变）唯一的ID，也即 Transaction ID。Transactin ID 与 PID 可能一一对应。区别在于 Transaction ID由用户提供，而 PID 是内部的实现对用户透明。

• 在 Producer 恢复时，每次 Producer 通过 Transaction ID 拿到 PID 的同时，还会获取一个单调递增的epoch。由于旧的Producer的epoch比新Producer的epoch小，Kafka可以很容易识别出该Producer是老的Producer并拒绝其请求。即维护一个epoch+一个版本号

有了 Transaction ID 后，Kafka 可保证：

• 跨 Session 的数据幂等发送。当具有相同Transaction ID 的新 Producer 实例被创建且工作时，旧的且拥有相同 Transaction ID 的 Producer 将不再工作。
• 跨 Session 的事务恢复。如果某个应用实例宕机，新的实例可以保证任何未完成的旧的事务要么 Commit 要么 Abort，使得新实例从一个正常状态开始工作。

Kafka 消费者

• consumer 采用 pull（拉）模式从 broker 中读取数据。
• push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由 broker 决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成 consumer 来不及处理消息，（广播与订阅）
• pull 模式则可以根据 consumer 的消费能力以适当的速率消费消息。
• pull 模式不足之处是，如果 kafka 没有数据，消费者可能会陷入循环中，一直返回空数据。针对这一点，Kafka 的消费者在消费数据时会传入一个时长参数 timeout，如果当前没有数据可供消费，consumer 会等待一段时间之后再返回，这段时长即为 timeout。
• 在消费者消费时，每个分区同一时间只能被同一消费组中的唯一一个消费者消费，但是多个消费者组可以同事消费同一个partition.

分区分配策略

一个 consumer group 中有多个 consumer，一个 topic 有多个 partition，所以必然会涉及到 partition 的分配问题，即确定那个 partition 由哪个 consumer 来消费。

• Kafka 有三种分配策略，RoundRobin、Range 以及 StickyAssignor

将分区的所有权从一个消费者移到另一个消费者称为重新平衡（rebalance）。当以下事件发生时，Kafka 将会进行一次分区分配：

• 同一个 Consumer Group 内新增消费者
• 消费者离开当前所属的 Consumer Group，包括 shuts down 或 crashe
• 订阅的主题新增分区

Range（默认策略）

Range 是对每个 Topic 而言的（即一个 Topic一个Topic分区）

• 首先对同一个Topic里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。
• 然后用 Partitions 分区的个数除以消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽，那么前面几个消费者线程将会多消费一个分区。
• 弊端： 一般topic 都无法除尽，也就是说增对于每个topic 都会存在一些消费者多消费一个分区。每个topic都会给这个消费者增加一个分区，N个topic，该消费者就多增加N个分区。由于编号了，所以每次都是固定的消费者多消费分区

RoundRobinAssignor

RoundRobinAssignor策略的原理是将消费组内所有消费者以及消费者所订阅的所有 topic的partition按照字典序排序，然后通过轮询消费者方式逐个将分区分配给每个消费者。。

• RoundRobinAssignor策略对应的partition.assignment.strategy参数值为：org.apache.kafka.clients.consumer.RoundRobinAssignor
• 如果同一个消费组内所有的消费者的订阅信息都是相同的，那么RoundRobinAssignor策略的分区分配会是均匀的。

举例，假设消费组中有 2 个消费者 C0 和 C1，都订阅了主题 t0 和 t1，并且每个主题都有3个分区，那么所订阅的所有分区可以标识为：t0p0、t0p1、t0p2、t1p0、t1p1、t1p2。最终的分配结果为：

如上图，左->右->左->右…如此反复轮训，如果两个消费者的消费能力相同，且订阅信息相同，那么就会被均匀分配。
弊端：

• 如果消费者订阅的Topic 不完全相同，如下
• 假设消费组内有3个消费者C0、C1和C2，它们共订阅了3个主题：t0、t1、t2，这3个主题分别有1、2、3个分区，即整个消费组订阅了t0p0、t1p0、t1p1、t2p0、t2p1、t2p2这6个分区。
  
  具体而言，消费者C0订阅的是主题t0，消费者C1订阅的是主题t0和t1，消费者C2订阅的是主题t0、t1和t2，那么最终的分配结果为：
  
• 可以看到RoundRobinAssignor策略也不是十分完美，这样分配其实并不是最优解，因为完全可以将分区t1p1分配给消费者C1，如下图：
  

StickyAssignor分配策略

我们再来看一下 StickyAssignor 策略，“sticky”这个单词可以翻译为“粘性的”，Kafka从0.11.x版本开始引入这种分配策略，它主要有两个目的：

• 分区的分配要尽可能的均匀；
• 分区的分配尽可能的与上次分配的保持相同。
• 当两者发生冲突时，第一个目标优先于第二个目标。

StickyAssignor的实现代码是RangeAssignor和RoundRobinAssignor的十倍，复杂度则远远在十倍以上。目前基本没有看到对这块源码实现的分析。我们举例来看一下StickyAssignor策略的实际效果。

1. 消费者订阅相同 Topic

• 假设消费组内有3个消费者：C0、C1和C2，它们都订阅了4个主题：t0、t1、t2、t3，并且每个主题有2个分区，也就是说整个消费组订阅了t0p0、t0p1、t1p0、t1p1、t2p0、t2p1、t3p0、t3p1这8个分区。最终的分配结果如下：
  
• 这样初看上去似乎与采用RoundRobinAssignor策略所分配的结果相同，但事实是否真的如此呢？再假设此时消费者C1脱离了消费组，那么消费组就会执行再平衡 *** 作，进而消费分区会重新分配。如果采用RoundRobinAssignor策略，那么此时的分配结果如下：
  
• 如分配结果所示，RoundRobinAssignor策略会按照消费者C0和C2进行重新轮询分配。而如果此时使用的是StickyAssignor策略，那么分配结果为：
  
• 可以看到分配结果中保留了上一次分配中对于消费者C0和C2的所有分配结果，并将原来消费者C1的“负担”分配给了剩余的两个消费者C0和C2，最终C0和C2的分配还保持了均衡。
• 简单的说就是粘性 *** 作，在发生重分配时，不进行全局的重分配，只将发生故障的消费者所对应的partition进行分配

2. 消费者订阅不同 Topic

举例，同样消费组内有3个消费者：C0、C1和C2，集群中有3个主题：t0、t1和t2，这3个主题分别有1、2、3个分区，也就是说集群中有t0p0、t1p0、t1p1、t2p0、t2p1、t2p2这6个分区。消费者C0订阅了主题t0，消费者C1订阅了主题t0和t1，消费者C2订阅了主题t0、t1和t2。

• 如果此时采用RoundRobinAssignor策略，那么最终的分配结果如下所示（和讲述RoundRobinAssignor策略时的一样，这样不妨赘述一下）：
  
• 如果此时采用的是StickyAssignor策略，那么最终的分配结果为：
  
• 消费者脱离消费组的情况 RoundRobin
  
• 而如果采用的是StickyAssignor策略，那么分配结果为：
  
• 可以看到StickyAssignor策略保留了消费者C1和C2中原有的5个分区的分配：t1p0、t1p1、t2p0、t2p1、t2p2。（针对结果集2, 保留了三个绿色的,结果集2如下图，做参照）。
  

Kafka 高效读写数据 的原因

1、顺序写磁盘：

• Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，为顺序写。
• 同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间

2、零复制技术：

• 传统OP流程：
  • 1、第一次：将磁盘文件，读取到 *** 作系统内核缓冲区；
  • 2、第二次：将内核缓冲区的数据，copy 到 application 应用程序的buffer；
  • 3、第三步：将application应用程序buffer中的数据，copy到socket网络发送缓冲区(属于 *** 作系统内核的缓冲区)；
  • 4、第四次：将socket buffer的数据，copy到网卡，由网卡进行网络传输。
    
• 零拷贝技术：尽可能的减少复制次数
  Customer从broker读取数据，采用sendfile，将磁盘文件读到OS内核缓冲区后，直接转到socket buffer进行网络发送。

3、memory Mapped files

• 将磁盘文件映射到内存, 用户通过修改内存就能修改磁盘文件。
• 它的工作原理是直接利用 *** 作系统的 Page 来实现文件到物理内存的直接映射。 完成映射之后，你对物理内存的 *** 作会被同步到硬盘上（ *** 作系统在适当的时候）。
• mmap也有一个很明显的缺陷——不可靠，写到mmap中的数据并没有被真正的写到硬盘， *** 作系统会在程序主动调用flush的时候才把数据真正的写到硬盘。

Zookeeper 在 Kafka 中的作用

• 配置管理
  Topic 的配置之所以能动态更新就是基于 zookeeper 做了一个动态全局配置管理。

• 负载均衡
  基于zookeeper 的消费者，实现了该特性，动态的感知分区变动，将负载使用既定策略分配到消费者身上。

• 命名服务
  Broker 将 advertised.port 和 advertised.host.name，这两个配置发布到zookeeper上的zookeeper的节点上/brokers/ids/BrokerId(broker.id),这个是供生产者，消费者，其它 Broker 跟其建立连接用的。

• 分布式通知
  比如分区增加，topic变动，Broker上线下线等均是基于zookeeper来实现的分布式通知。

• 集群管理和master选举
  我们可以在通过命令行，对 kafka 集群上的 topic partition 分布，进行迁移管理，也可以对 partition leader 选举进行干预。
  Master选举，要说有也是违反常规，常规的master选举，是基于临时顺序节点来实现的，序列号最小的作为master。而kafka的Controller的选举是基于临时节点来实现的，临时节点创建成功的成为Controller，更像一个独占锁服务。

• 分布式锁
  独占锁，用于 Controller 的选举。