震惊了，原来这才是Kafka的“真面目”！

2020-05-26 00:00:00 消息 事务 消费 消费者 记录


导读：Kafka 是一个分布式消息队列，具有高性能、持久化、多副本备份、横向扩展能力。生产者往队列里写消息，消费者从队列里取消息进行业务逻辑。一般在架构设计中起到解耦、削峰、异步处理的作用。



Kafka 对外使用 Topic 的概念，生产者往 Topic 里写消息，消费者从中读消息。为了做到水平扩展，一个 Topic 实际是由多个 Partition 组成的，遇到瓶颈时，可以通过增加 Partition 的数量来进行横向扩容。单个 Parition 内是保证消息有序。

每新写一条消息，Kafka 就是在对应的文件 append 写，所以性能非常高。

Kafka 的总体数据流是这样的：


大概用法就是，Producers 往 Brokers 里面的指定 Topic 中写消息，Consumers 从 Brokers 里面拉取指定 Topic 的消息，然后进行业务处理。图中有两个 Topic，Topic0 有两个 Partition，Topic1 有一个 Partition，三副本备份。


可以看到 Consumer Gourp1 中的 Consumer2 没有分到 Partition 处理，这是有可能出现的，下面会讲到。关于 Broker、Topics、Partitions 的一些元信息用 ZK 来存，监控和路由啥的也都会用到 ZK。




生产


基本流程是这样的：


创建一条记录，记录中一个要指定对应的 Topic 和 Value，Key 和 Partition 可选。


先序列化，然后按照 Topic 和 Partition，放进对应的发送队列中。Kafka Produce 都是批量请求，会积攒一批，然后一起发送，不是调 send（）就立刻进行网络发包。


如果 Partition 没填，那么情况会是这样的：

  • Key 有填。按照 Key 进行哈希，相同 Key 去一个 Partition。（如果扩展了 Partition 的数量那么就不能保证了）
  • Key 没填。Round-Robin 来选 Partition。


这些要发往同一个 Partition 的请求按照配置，攒一波，然后由一个单独的线程一次性发过去。


API


有 High Level API，替我们把很多事情都干了，Offset，路由啥都替我们干了，用起来很简单。还有 Simple API，Offset 啥的都是要我们自己记录。（注：消息消费的时候，首先要知道去哪消费，这就是路由，消费完之后，要记录消费单哪，就是 Offset）


Partition


当存在多副本的情况下，会尽量把多个副本，分配到不同的 Broker 上。Kafka 会为 Partition 选出一个 Leader，之后所有该 Partition 的请求，实际操作的都是 Leader，然后再同步到其他的 Follower。


当一个 Broker 歇菜后，所有 Leader 在该 Broker 上的 Partition 都会重新选举，选出一个 Leader。（这里不像分布式文件存储系统那样会自动进行复制保持副本数）


然后这里就涉及两个细节：

  • 怎么分配 Partition
  • 怎么选 Leader


关于 Partition 的分配，还有 Leader 的选举，总得有个执行者。在 Kafka 中，这个执行者就叫 Controller。Kafka 使用 ZK 在 Broker 中选出一个 Controller，用于 Partition 分配和 Leader 选举。


Partition 的分配：


  • 将所有 Broker（假设共 n 个 Broker）和待分配的 Partition 排序。
  • 将第 i 个 Partition 分配到第（i mod n）个 Broker 上 （这个就是 Leader）。
  • 将第 i 个 Partition 的第 j 个 Replica 分配到第（(i + j) mode n）个 Broker 上。


Leader 容灾


Controller 会在 ZK 的 /brokers/ids 节点上注册 Watch，一旦有 Broker 宕机，它就能知道。当 Broker 宕机后，Controller 就会给受到影响的 Partition 选出新 Leader。


Controller 从 ZK 的 /brokers/topics/[topic]/partitions/[partition]/state 中，读取对应 Partition 的 ISR（in-sync replica 已同步的副本）列表，选一个出来做 Leader。选出 Leader后，更新ZK，然后发送 LeaderAndISRRequest 给受影响的 Broker，让它们知道改变这事。


为什么这里不是使用 ZK 通知，而是直接给 Broker 发送 RPC 请求，我的理解可能是这样做 ZK 有性能问题吧。如果 ISR 列表是空，那么会根据配置，随便选一个 Replica 做 Leader，或者干脆这个 Partition 就是歇菜；如果 ISR 列表的有机器，但是也歇菜了，那么还可以等 ISR 的机器活过来。


多副本同步


这里的策略，服务端这边的处理是 Follower 从 Leader 批量拉取数据来同步。但是具体的可靠性，是由生产者来决定的。生产者生产消息的时候，通过 request.required.acks 参数来设置数据的可靠性。


在 Acks=-1 的时候，如果 ISR 少于 min.insync.replicas 指定的数目，那么就会返回不可用。


这里 ISR 列表中的机器是会变化的，根据配置 replica.lag.time.max.ms，多久没同步，就会从 ISR 列表中剔除。以前还有根据落后多少条消息就踢出 ISR，在 1.0 版本后就去掉了，因为这个值很难取，在高峰的时候很容易出现节点不断的进出 ISR 列表。从 ISA 中选出 Leader 后，Follower 会把自己日志中上一个高水位后面的记录去掉，然后去和 Leader 拿新的数据。


因为新的 Leader 选出来后，Follower 上面的数据，可能比新 Leader 多，所以要截取。这里高水位的意思，对于 Partition 和 Leader，就是所有 ISR 中都有的新一条记录。消费者多只能读到高水位。

从 Leader 的角度来说高水位的更新会延迟一轮，例如写入了一条新消息，ISR 中的 Broker 都 Fetch 到了，但是 ISR 中的 Broker 只有在下一轮的 Fetch 中才能告诉 Leader。也正是由于这个高水位延迟一轮，在一些情况下，Kafka 会出现丢数据和主备数据不一致的情况，0.11 开始，使用 Leader Epoch 来代替高水位。


思考：当 Acks=-1 时

  • 是 Follwers 都来 Fetch 就返回成功，还是等 Follwers 第二轮 Fetch？
  • Leader 已经写入本地，但是 ISR 中有些机器失败，那么怎么处理呢？




消费


订阅 Topic 是以一个消费组来订阅的，一个消费组里面可以有多个消费者。同一个消费组中的两个消费者，不会同时消费一个 Partition。换句话来说，就是一个 Partition，只能被消费组里的一个消费者消费，但是可以同时被多个消费组消费。


因此，如果消费组内的消费者如果比 Partition 多的话，那么就会有个别消费者一直空闲。


API


订阅 Topic 时，可以用正则表达式，如果有新 Topic 匹配上，那能自动订阅上。


Offset 的保存


一个消费组消费 Partition，需要保存 Offset 记录消费到哪，以前保存在 ZK 中，由于 ZK 的写性能不好，以前的解决方法都是 Consumer 每隔一分钟上报一次。这里 ZK 的性能严重影响了消费的速度，而且很容易出现重复消费。在 0.10 版本后，Kafka 把这个 Offset 的保存，从 ZK 总剥离，保存在一个名叫 consumeroffsets topic 的 Topic 中。

写进消息的 Key 由 Groupid、Topic、Partition 组成，Value 是偏移量 Offset。Topic 配置的清理策略是 Compact。总是保留新的 Key，其余删掉。一般情况下，每个 Key 的 Offset 都是缓存在内存中，查询的时候不用遍历 Partition，如果没有缓存，次就会遍历 Partition 建立缓存，然后查询返回。


确定 Consumer Group 位移信息写入 consumers_offsets 的哪个 Partition，具体计算公式：

__consumers_offsets partition =
           Math.abs(groupId.hashCode() % groupMetadataTopicPartitionCount)   
//groupMetadataTopicPartitionCount由offsets.topic.num.partitions指定，默认是50个分区。

相关文章