消息引擎系统

定义

消息格式

传输协议

好处

KAFKA术语

• 消息(record) — kafka处理的对象

• 主题(topic) — 发布订阅的对象，承载消息的逻辑容器

• 生产者(producer) — 向主题发布消息的客户端应用程序

• 消费者(consumer) — 订阅主题消息的客户端应用程序 

• 服务器端(broker) — 接收和处理客户端请求，对消息进行持久化

• 分区(partition) — 一个有序不变的消息序列。一topic多分区，消息只在某一个分区 

• 备份(replication) — 相同的数据拷贝在多台机器上 

• 副本(replica) — 相同的数据拷贝，冗余实现高可用，副本是分区维度的副本，分区内配置若干副本，领导者副本对外提供服务，追随者副本不与外界交互

• 消息位移(offset) — 分区中每条消息的位置信息

• 消费者位移(consumer offset) — 消费者消费进度

• 消费者组(consumer group) — 多个消费者实例共同组成的一个组，同时消费多个分区以实现高吞吐，每个分区都只会被组内一个消费者实例消费

• 重平衡(rebalance) — 消费者组内新增实例or某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程

• 生产者->领导者副本<-消费者

KAFKA版本号

• 0.7 上古版本，只提供最基础的消息队列功能，没有副本机制

• 0.8 引入副本机制，老版本客户端api(指定zookeeper地址而不是broker地址)

• 0.8.2.0 引入新版producer api(不稳定)

• 0.8.2.2 老版消费者api较为稳定

• 0.9 新版producer api较为稳定，引入新版consumer api(不稳定)

• 0.10 引入kafka streams，对消息引擎方面没有太大功能提升

• 0.10.2.2 新版consumer api较为稳定

• 0.11 引入提供幂等性producer api和事务api(不稳定)，重构消息格式

• 0.11.0.3 消息引擎功能较为完善

• 1.0 and 2.0 kafka streams各种改进

• 公司kafka版本0.10.1.2

分区

作用

1. 提供负载均衡能力，实现系统的高伸缩性。不同的分区能够被放置到不同节点的机器上，还可以通过添加新的节点机器来增加整体系统的吞吐量 

2. 实现业务级别的消息顺序问题。将一组特定有序的消息发到同一分区，保证顺序处理。

分区策略

轮询策略

随机策略

消息键保序策略

可靠性保障

至少一次(at least once)

至多一次(at most once)

精确一次(exactly once)

消费者组

• Consumer Group下可以有一个或多个Consumer实例。这里的实例可以是一个单独的进程，也可以是同一进程下的线程。在实际场景中，使用进程更为常见一些。 

• Group ID是一个字符串，在一个Kafka集群中，它标识唯一的一个Consumer Group。 

• Consumer Group下所有实例订阅的主题的单个分区，只能分配给组内的某个Consumer实例消费。这个分区当然也可以被其他的Group消费。 

• 消费者位移，老版zookeeper，由于不适合频繁写操作，新版保存在broker的内部主题(位移主题)中。key<Group ID，主题名，分区号> ，value有三种格式(位移值、保存消费者组信息，注册时使用、删除消费者组过期位移甚至是删除消费者组)

重平衡

定义

触发原因

1. 组成员数发生变更。比如有新的Consumer实例加入组或者离开组，或是有Consumer实例崩溃被“踢 出”组。

2. 订阅主题数发生变更。Consumer Group可以使用正则表达式的方式订阅主题。在运行过程中，新创建了一个满足条件的主题，那么该Group就会发生Rebalance。 

3. 订阅主题的分区数发生变更。当分区数增加时，就会触发订阅该主题的所有Group开启Rebalance。 

消费者端流程

加入组(JoinGroup)

1. 每个成员将自己订阅的主题上报

2. 协调者从中选择一个成员担任消费者组的领导者（第一个JoinGroup的消费者）

3. 协调者把消费者组订阅信息封装进JoinGroup请求的响应体中发给领导者

4. 领导者做出分配方案

等待领导者消费者分配方案(SyncGroup)

1. 领导者向协调者发送SyncGroup请求，将分配方案发送给协调者（其他消费者也会发送该请求，请求体内没有实际内容）

2. 协调者以SyncGroup响应的方式将分配方案分发给所有消费者成员

协调者端流程

新成员入组（Stable状态时，有新成员加入）

组成员主动离组（主动调用close）

组成员崩溃离组

重平衡时协调者对组内成员提交位移的处理

影响

如何规避

• session.timeout.ms，接收心跳最大时长，默认10s

• heartbeat.interval.ms，发送心跳的频率，默认3s

• max.poll.interval.ms，消费消息最大时长，默认5分钟

• max.poll.records，下游一次消费的消息数，默认500条

1. 第一类非必要Rebalance是因为未能及时发送心跳，导致Consumer被“踢出”Group而引发的。比较合理的配置为能保证Consumer实例在被判定为“dead”之前，能够发送至少3轮的心跳请求。

2. 第二类非必要Rebalance是Consumer消费时间过长导致的。设置max.poll.interval.ms要大于可能处理的最大时长，减小max.poll.records数量。

3. consumer消费时有错误导致重启触发rebalance

消费者位移提交

消费者TCP管理

何时连接

1. FindCoordinator请求，确定协调者和获取集群元数据。

2. 连接协调者，令其执行组成员管理操作。

3. 连接分区领导者副本对应broker，执行实际的消息获取。

何时关闭

副本机制

• In-sync Replicas（ISR副本集合）落后leader副本小于replica.lag.time.max.ms的副本集合，kafka认为该集合中的副本是与leader同步的。

• Unclean领导者选举（是否允许非同步副本被选举为leader）

broker处理请求流程

控制器组件（Controller）

控制器是如何被选出来的

控制器是做什么的

主题管理（创建、删除、增加分区）

分区重分配

Preferred领导者选举

集群成员管理（自动检测新增Broker、Broker主动关闭、Broker宕机）

数据服务

控制器保存了什么数据

控制器故障转移（Failover）

1. Broker 0宕机

2. ZooKeeper通过Watch机制感知并删除/controller临时节点

3. 所有存活的Broker竞选新的控制器身份

4. Broker 3赢得了选举，在ZooKeeper重建了/controller节点

5. Broker 3从ZooKeeper中读取集群元数据信息，并初始化到自己的缓存中

副本同步机制

什么是高水位

水位

• 在时刻T，任意创建时间（Event Time）为T’，且T’≤T的所有事件都已经到达或被观测到，那么T就被定义为水位。 -- 经典定义

• 水位是一个单调增加且表征最早未完成工作（oldest work not yet completed）的时间戳。                -- Streaming System

KAFKA水位

• 在分区高水位以下的消息被认为是已提交消息，反之就是未提交消息。消费者只能消费已提交消息，即图中位移小于8的所有消息。

• 位移值等于高水位的消息也属于未提交消息。也就是说，高水位上的消息是不能被消费者消费的。图中还有一个日志末端位移(Log End Offset)的概念，简写是LEO。它表示副本写入下一条消息的位移值。介于高水位和LEO之间的消息就属于未提交消息。同一个副本对象，其高水位值不会大于LEO值。

• 高水位和LEO是副本对象的两个重要属性。Kafka所有副本都有对应的高水位和LEO值，而分区的高水位就是其Leader副本的高水位。

高水位的作用

1. 定义消息可见性，即用来标识分区下的哪些消息是可以被消费者消费的。

2. 帮助Kafka完成副本同步。

高水位更新机制

Leader副本更新机制

1. 写入消息到本地磁盘并更新LEO值。

2. 获取Leader副本所在Broker保存的所有与Leader同步的远程副本LEO值{LEO-1，LEO-2，……，LEO-n}。

3. 更新高水位为min(LEO, LEO-1，LEO-2，……，LEO-n)。

1. 读取磁盘（或页缓存）中的消息数据。

2. 使用Follower副本发送请求中的LEO值更新远程副本LEO值。

3. 更新分区高水位值（具体步骤与处理生产者请求的步骤相同）。

Follower副本更新机制

1. 写入消息到本地磁盘并更新LEO值。

2. 更新高水位为min(LeaderHW, currentLEO)。

1. 该远程Follower副本在ISR中。

2. 该远程Follower副本LEO值落后于Leader副本LEO值的时间，不超过replica.lag.time.max.ms（默认10s）。

副本同步机制解析

• 在初始状态时，所有值都是0

• 当生产者给主题分区发送一条消息后，状态变更为：

• Follower再次尝试从Leader拉取消息，状态进一步变更为： 

• 在新一轮的拉取请求中，Follower副本请求拉取位移值为1的消息。

• Leader副本收到请求后，更新远程副本LEO为1，然后更新Leader高水位为1。最后将高水位值1发送给Follower副本。

• Follower副本接收到以后，也将自己的高水位值更新成1。

Leader Epoch

上述流程存在的问题

• 开始时，副本A和副本B都处于正常状态，A是Leader副本。某个使用了min.insync.replicas=1（Leader副本成功写入即算发送成功）的生产者程序向A发送了两条消息，A全部写入成功，此时Kafka会通知生产者两条消息全部发送成功。

• 现在假设Leader和Follower都写入了这两条消息，而且Leader副本的高水位也已经更新了，但Follower副本高水位还未更新。

• 倘若此时副本B所在的Broker宕机，当它重启回来后，副本B会执行日志截断操作，将LEO值调整为之前的高水位值，也就是1。也就是位移值为1的那条消息被副本B从磁盘中删除，此时副本B的磁盘中只保存了位移值为0的那条消息。

• 执行完截断操作后，副本B开始从A拉取消息，执行正常的消息同步。如果在此时，副本A所在的Broker宕机了，那么Kafka就别无选择，只能让副本B成为新的Leader。

• 当A回来后，需要执行相同的日志截断操作，即将高水位调整为与B相同的值，也就是1。

• 这样操作之后，位移值为1的那条消息就从这两个副本中被永远地抹掉了。这就是这张图要展示的数据丢失场景。

数据不一致

• 这种情况的初始状态与上面有一些不同：A依然是Leader，写入了2条消息，min.insync.replicas依旧为1，但B只写入了1条消息且不在ISR中。分区HW更新到2，但B的HW还是1。

• 这次我们让A和B所在机器同时挂掉，然后假设B先重启回来，假设允许Unclean领导者选举，B因此成为leader，分区HW=1。假设此时producer发送了第3条消息(绿色框表示)给B，于是B的log中offset=1的消息变成了绿色框表示的消息，同时分区HW更新到2（A还没有回来，就B一个副本，故可以直接更新HW而不用理会A）。之后A重启回来，发现此时分区HW=2，与自己的HW值相同，故不做任何调整。此后A和B将以这种状态继续正常工作。

• 这种场景下，A和B在offset=1的消息是不同的记录，从而导致数据不一致。

介绍

1. Epoch。一个单调增加的版本号。每当副本领导权发生变更时，都会增加该版本号。小版本号的Leader被认为是过期Leader，不能再行使Leader权力。

2. 起始位移（Start Offset）。Leader副本在该Epoch值上写入的首条消息的位移。

• 举例来说明一下Leader Epoch。假设现在有两个Leader Epoch<0, 0>和<1, 120>，那么，第一个Leader Epoch表示版本号是0，这个版本的Leader从位移0开始保存消息，一共保存了120条消息。之后Leader发生了变更，版本号增加到1，新版本的起始位移是120。

• Kafka Broker会在内存中为每个分区都缓存Leader Epoch数据，同时定期持久化到checkpoint文件中。Leader副本写入消息到磁盘时，Broker会尝试更新这部分缓存。当发生Leader变更时，新的Leader副本会查询这部分缓存，取出对应的Leader Epoch的起始位移，以避免数据丢失和不一致的情况。

• OffsetsForLeaderEpochRequest请求：副本重启后，会读取checkpoint文件获取当前的Epoch版本，并向Leader副本请求下一版本的起始Offset。若当前Leader副本的Epoch版本与该副本相同，则返回当前LEO值。若相差版本较多，恢复的副本会一步步把自己和Leader之间缺少的版本信息和数据补齐。

问题解决

数据丢失

• 场景和之前大致类似，只不过副本B重启后，从checkpoint文件中获取Leader Epoch，得知自己的Epoch是0，需要向A发送OffsetsForLeaderEpochRequest请求由于A的版本也是0，返回当前A的LEO值，该值为2。当获知到Leader LEO=2后，B发现该LEO值与自己的LEO值相同，因此B无需执行日志截断操作。这是对高水位机制的一个明显改进，即副本是否执行日志截断不再依赖于高水位进行判断。

• 现在，副本A宕机，B成为Leader。A重启后，执行与B相同的逻辑判断，发现也不用执行日志截断，至此位移值为1的那条消息在两个副本中均得到保留。后面当生产者向B写入新消息时，副本B所在的Broker缓存中，会生成新的Leader Epoch条目：[Epoch=1, Offset=2]。之后，副本B会使用这个条目帮助判断后续是否执行日志截断操作。

数据不一致