整体架构

核心概念

消息

• Kafka中最基本的数据单元
• 消息是一串主要由key和value构成的字符串，key和value也都是byte数组
• key的主要作用是根据一定的策略，将此消息路由到指定的Partition中，这样就可以保证包含同一key的消息全部写入同一分区中
• 消息的真正有效负载是value部分的数据
• 为了提高网络和存储的利用率，Producer会批量发送消息到Kafka，并在发送之前对消息进行压缩。

Producer

• 负责将消息发送到Kafka集群，即将消息按照一定的规则推送到Topic的Partition中
• 选择分区的“规则”可以有很多种
  • 根据消息的key的Hash值选择Partition
  • 按序轮训该Topic全部Partition的方式

Broker

• Kafka集群中一个单独的Kafka Server就是一个Broker
• Broker的主要工作就是接收Producer发过来的消息、为其分配offset并将消息保存到磁盘中
• 同时，接收Consumer以及其他Broker的请求，并根据请求类型进行相应处理并返回响应。

Topic

• Topic是用于存储消息的逻辑概念，可以看作是一个消息集合
• 发送到Kafka集群的每条消息都存储到一个Topic中
• 每个Topic可以有多个生产者向其中推送（push）消息，也可以有任意多个消费者消费其中的消息

Partition

• 每个Topic可以划分成一个或多个Partition，同一Topic下的不同分区包含着消息是不同的。
• 每个消息在被添加到Partition时，都会被分配一个offset，它是消息在此分区中的唯一编号，Kafka通过offset保证消息在分区内的顺序
• offset的顺序性不跨分区，即Kafka只保证在同一个分区内的消息是有序的；同一Topic的多个分区内的消息，Kafka并不保证其顺序性
  

Log

• 同一Topic的不同 Partition 会分配在不同的Broker上。 Partition是Kafka水平扩展性的基础，我们可以通过增加服务器并在其上分配Partition的方式，增加Kafka的并行处理能力。
• Partition在逻辑上对应着一个Log，当Producer将消息写入Partition时，实际上是写入到了Partition对应的Log中
• Log 是一个逻辑概念，可以对应到磁盘上的一个文件夹
• Log 由多个Segment组成，每个Segment对应一个日志文件和索引文件
• 在面对海量数据时，为避免出现超大文件，每个日志文件的大小是有限制的，当超出限制后则会创建新的 Segment，继续对外提供服务。这里要注意，因为Kafka采用顺序IO，所以只向最新的Segment追加数据。为了权衡文件大小、索引速度、占用内存大小等多方面因素，索引文件采用稀疏索引的方式，文件大小并不会很大，在运行时会将其内容映射到内存，提高索引速度。

保留策略（Retention Policy）

• 无论消费者是否已经消费了消息，Kafka都会一直保存这些消息，但并不会像数据库那样长期保存。为了避免磁盘被占满，Kafka会配置相应的“保留策略”（Retention Policy），以实现周期性的删除陈旧的消息。
• Kafka 中有两种“保留策略”
  • 一种是根据消息保留的时间，当消息在 Kafka 中保存的时间超过了指定时间，就可以被删除
  • 另一种是根据 Topic 存储的数据大小，当 Topic 所占的日志文件大小大于一个阈值，则可以开始删除最旧的消息
• Kafka 会启动一个后台线程，定期检查是否存在可以删除的消息。“保留策略”的配置是非常灵活的，可以有全局的配置，也可以针对 Topic 进行配置覆盖全局配置。

日志压缩（Log Compaction）

• 在很多场景中，消息的key与value的值之间的对应关系是不断变化的，就像数据库中的数据会不断被修改一样，消费者只关心 key 对应的最新 value 值
• 此时，可以开启Kafka的日志压缩功能，Kafka会在后台启动一个线程，定期将相同key的消息进行合并，只保留最新的value值。
  

Replica副本

• 一般情况下，Kafka 对消息进行了冗余备份，每个 Partition 可以有多个 Replica（副本），每个 Replica 中包含的消息是一样的
• 每个 Partition 的Replica集合中，都会选举出一个Replica作为Leader Replica，Kafka在不同的场景下会采用不同的选举策略
• 所有的读写请求都由选举出的 Leader Replica 处理，其他都作为Follower Replica，Follower Replica仅仅是从Leader Replica处把数据拉取到本地之后，同步更新到自己的 Log 中
• 每个 Partition 至少有一个 Replica，当 Partition 中只有一个 Replica 时，就只有 Leader Replica，没有 Follower Replica
• 一般情况下，同一 Partition 的多个 Replica 会被分配到不同的 Broker 上，这样，当 Leader 所在的 Broker 宕机之后，可以重新选举新的 Leader，继续对外提供服务。
  

ISR 集合

• ISR（In-Sync Replica）集合表示的是目前“可用”（alive）且消息量与 Leader 相差不多的副本集合，这是整个副本集合的一个子集。
• “可用”和“相差不多”都是很模糊的描述，其实际含义是ISR集合中的副本必须满足下面两个条件：
  • 副本所在节点必须维持着与ZooKeeper的连接。
  • 副本最后一条消息的offset与Leader副本的最后一条消息的offset之间的差值不能超出指定的阈值。
• 每个分区中的 Leader Replica 都会维护此分区的 ISR 集合。
• 写请求首先是由 Leader Replica 处理，之后 Follower Replica 会从 Leader Replica 上拉取写入的消息，这个过程会有一定的延迟，导致 Follower Replica 中保存的消息略少于 Leader Replica，只要未超出阈值都是可以容忍的。
• 如果一个 Follower Replica 出现异常，比如：宕机、发生长时间 GC 而导致 Kafka 僵死或是网络断开连接导致长时间没有拉取消息进行同步，就会违反上面的两个条件，从而被 Leader Replica 踢出 ISR 集合
• 当 Follower Replica 从异常中恢复之后，会继续与 Leader Replica 进行同步，当 Follower Replica “追上” Leader Replica 的时候（即最后一条消息的 offset 的差值小于指定阈值），此 Follower Replica 会被 Leader Replica 重新加入 ISR 集合中。

HW

• HW 标记了一个特殊的 offset ，当消费者处理消息的时候，只能拉取到 HW 之前的消息，HW 之后的消息对消费者来说是不可见的。
• 与 ISR 集合类似，HW 也是由 Leader  Replica管理的。当ISR集合中全部的Follower Replica 都拉取 HW 指定消息进行同步后，Leader Replica 会递增 HW 的值。
• Kafka 官方网站的将 HW 之前的消息的状态称为“commit”，其含义是这些消息在多个 Replica 中同时存在，即使此时 Leader Replica 损坏，也不会出现数据丢失。

LEO

• LEO（Log End offset）是所有的Replica都会有的一个offset标记，它指向追加到当前Replica的最后一个消息的 offset 。
• 当 Producer 向 Leader Replica 追加消息的时候， Leader Replica的LEO标记会递增；当Follower Replica 成功从Leader Replica拉取消息并更新到本地的时候，Follower Replica 的 LEO 就会增加。

1. Producer 向此 Partition 推送消息。
2. Leader Replica 将消息追加到 Log 中，并递增其 LEO。
3. Follower Replica 从 Leader Replica 拉取消息进行同步。
4. Follower Replica 将拉取到的消息更新到本地 Log 中，并递增其 LEO 。
5. 当 ISR 集合中所有 Replica 都完成了对 offset=11的消息的同步，Leader Replica会递增HW。

• 在 ①~⑤ 步完成之后，offset=11 的消息就对 Consumer 可见了。

Cluster&Controller

• 多个 Broker 可以做成一个 Cluster（集群）对外提供服务
• 每个Cluster当中会选举出一个Broker来担任Controller，Controller是Kafka集群的指挥中心，而其他 Broker 则听从 Controller 指挥实现相应的功能
• Controller负责管理分区的状态、管理每个分区的Replica状态、监听Zookeeper中数据的变化等工作。
• Controller 也是一主多从的实现，所有 Broker 都会监听 Controller Leader 的状态，当 Leader Controller 出现故障时则重新选举新的 Controller Leader。

Consumer

• 从 Topic 中拉取消息，并对消息进行消费。
• 某个消费者消费到 Partition 的哪个位置（offset）的相关信息，是 Consumer 自己维护的。
  
• 这样设计非常巧妙，避免了Kafka Server端维护消费者消费位置的开销，尤其是在消费数量较多的情况下。
• 另一方面，如果是由 Kafka Server 端管理每个 Consumer 消费状态，一旦 Kafka Server 端出现延或是消费状态丢失时，将会影响大量的 Consumer。
• 这一设计也提高了 Consumer的灵活性，Consumer可以按照自己需要的顺序和模式拉取消息进行消费。
  • 例如：Consumer可以通过修改其消费的位置实现针对某些特殊key的消息进行反复消费，或是跳过某些消息的需求。

Consumer Group

• 在 Kafka中，多个Consumer可以组成一个ConsumerGroup，一个Consumer只能属于一个Consumer Group
• Consumer Group保证其订阅的Topic的每个Partition只被分配给此Consumer Group中的一个消费者处理
• 如果不同 Consumer Group 订阅了同一 Topic，Consumer Group 彼此之间不会干扰。
  • 要实现一个消息可以被多个 Consumer 同时消费（“广播”）的效果;则将每个 Consumer 放入单独的一个 Consumer Group；
  • 如果要实现一个消息只被一个 Consumer 消费（“独占”）的效果，则将所有的 Consumer 放入一个 Consumer Group 中。
• 在 Kafka 官网的介绍中，将 Consumer Group 称为“逻辑上的订阅者”（logical subscriber）
  
• Consumer Group 除了实现“独占”和“广播”模式的消息处理外，Kafka 还通过 Consumer Group 实现了消费者的水平扩展和故障转移。
• 当 Consumer3 的处理能力不足以处理两个 Partition 中的数据时，可以通过向 Consumer Group 中添加消费者的方式，触发Rebalance 操作重新分配 Partition 与 Consumer 的对应关系，从而实现水平扩展。
  
• Consumer 出现故障的场景，当 Consumer4 宕机时，Consumer Group 会自动重新分配 Partition
  
• 注意，Consumer Group 中消费者的数量并不是越多越好，当消费者数量超过 Partition 的数量时，会导致有 Consumer 分配不到 Partition，从而造成 Consumer 的浪费。

设计原则

ISR&Leader Replica&Follower Replica

• 为什么Kafka要这么设计？在分布式存储中，冗余备份是常见的一种设计，常用的方案有同步复制和异步复制：
  • 同步复制要求所有能工作的Follower Replica都复制完，这条消息才会被认为提交成功。一旦有一个 Follower Replica出现故障，就会导致HW无法完成递增，消息就无法提交，消费者获取不到消息。这种情况下，故障的 Follower Replica 会拖慢整个系统的性能，甚至导致整个系统不可用。
  • 异步复制中，Leader Replica 收到生产者推送的消息后，就认为此消息提交成功。 Follower Replica 则异步地从 Leader Replica同步消息。这种设计虽然避免了同步复制的问题，但同样也存在一定的风险，现在假设所有Follower Replica的同步速度都比较慢，它们保存的消息量都远远落后于 Leader Replica。

• 异步复制下，此时Leader Replica所在的Broker突然宕机，则会重新选举新的Leader Replica，而新的 Leader Replica 中没有原来 Leader Replica 的消息，这就出现了消息的丢失，而有些 Consumer 则可能消费了这些丢失的消息，后续服务状态变得不可控。

• Kafka权衡了同步复制和异步复制两种策略，通过引入了ISR集合，巧妙地解决了上面两种方案存在的缺陷

  • 首先，当 Follower Replica 的延迟过高时，会将 Leader Replica 被踢出ISR集合，消息依然可以快速提交，Producer也可以快速得到响应，避免高延时的Follower Replica影响整个Kafka集群的性能。
  • 当Leader Replica所在的Broker突然宕机的时候，会优先将ISR集合中Follower Replica选举为 Leader Replica，新Leader  Replica中包含了HW之前的全部消息，这就避免了消息的丢失。
  • 值得注意是，Follower Replica 可以批量地从 Leader Replica复制消息，这就加快了网络 I/O，Follower Replica 在更新消息时是批量写磁盘，加速了磁盘的I/O，极大减少了Follower 与 Leader 的差距。

总结

• Producer 会根据业务逻辑产生消息，之后根据路由规则将消息发送到指定的分区的 Leader Replica 所在的 Broker 上。
• 在 Kafka 服务端接收到消息后，会将消息追加到 Leader Replica 的 Log 中保存，之后 Follower Replica 会与 Leader Replica 进行同步，当 ISR 集合中所有 Replica 都完成了此消息的同步之后，则 Leader Replica 的 HW 会增加，并向 Producer 返回响应。
• 当 Consumer 加入 Consumer Group 时，会触发 Rebalance 操作将 Partition 分配给不同的 Consumer 进行消费。
• 随后，Consumer 会确定其消费的位置，并向 Kafka 集群发送拉取消息的请求，Leader Replica会验证请求的offset以及其他相关信息，然后批量返回消息。