Kafka消息队列理解

小灰灰 2021-10-13 00:23 459阅读 0赞

### 什么是Kafka ###

Apache Kafka是一个基于发布/订阅的消息系统.

### Kafka的核心概念 ###

**消息**

Kafka的内部数据单元称为消息, 对于Kafka来说, 消息只是一个字节数组, 它不关心数据的格式. 消息有三个核心属性, 分别是offset, key, value. 偏移量(offset)是一个持续增长的整数值, 分区内部的每个消息都具有唯一的偏移量, 代表在分区内的位置, 或者说顺序. key除了作为一个标识外, 还用来控制消息被分派到哪个分区（partition）中. 最简单的做法是对key进行哈希, 把得到的哈希值对分区数量进行取模, 最后得到的取模值决定了消息分派到哪个分区. 这样可以保证拥有相同key的消息被写入同一个分区. 如果没有设置key, 则轮询分配消息到分区.

为了更高效, 消息以批量提交的方式来写入Kafka, 而且批量提交的消息是经过压缩的. 这些批量的消息都拥有相同的主题（topic）和分区（partition）. 批量提交消息是吞吐和延迟的一个权衡, 批量提交的消息越多，单位时间可以处理的消息也就越多，但每个消息的延迟就越大.

**主题（topic）与分区（partition）**

每条提交到Kafka集群的消息都有一个类别，这个类别被称为主题（topic）. 分区（partition）是物理上的概念, 每个主题包含一个或多个分区, 每个分区在物理上对应为一个文件目录，其命名规则为$\{topicName\}-\{partitionId\}. 分区目录下存储的是三个日志文件, 包括一个数据文件和两个索引文件(偏移量索引文件和时间戳索引文件), 消息被追加到相应的分区日志文件中, 是顺序写磁盘, 因此效率非常高, 这也是Kafka高吞吐率的一个重要保证. Kafka保证在一个分区内的消息是有序的, 但是不保证跨分区的全局顺序, 如果需要全局消息有序, 主题只能有一个分区.

> 日志文件按照大小或者时间滚动切分成一个或者多个日志段(LogSegment), 其中日志段大小由配置项log.segment.bytes指定, 默认是1GB. 时间长度则是根据log.roll.ms或者log.roll.hours配置项设置, 当日志段消息量或保留时间超过配置就会被删除. 正在写入的日志段称为称之为活跃段(activeSegment), 活跃段不会被删除.

**生产者与消费者**

Kafka客户端有两种基本类型: 生产者（producer）和消费者（consumer）.

生产者（也称为发布者）创建消息, 通常来说生产者不关心消息被写入到哪个分区，在默认情况下Kafka会将消息均衡分配到所有的分区(轮询的方式), 但如果我们想控制消息的分配, 我们可以设置键值（key）和分配器（partitioner）, 默认分配器是对key进行哈希, 然后再对分区数进行取模, 我们也可以自定义分配器.

消费者（也称为订阅者）读取消息, 通常订阅一个或多个主题. 在读取时，消费者会读取到消息在分区内的偏移量（offset）, 当消费者处理完消息后提交结果给Kafka, Kafka会记录这个偏移量, 以避免消息重复消费.

> 在老版本偏移量记录在Zookeeper中, 在新版本偏移量记录在Kafka分区日志文件中.

**为什么消息要有一个偏移量呢?**

因为消费者读取消息是一个fetch的操作, 而不是push的操作, 如果是push操作, kafka会主动将消息push给消费者, 这种方式就不需要偏移量. 但如果是fetch操作, 为了避免重复消费, 消息在分区中就需要有一个偏移量, 且Kafka还需要记录分区消费到了哪个偏移量. 不适用push模式的原因是, push模式很难适应消费速率不同的消费者, push模式的目标是尽可能以最快速度传递消息, 但是这样很容易造成消费者来不及处理消息, 典型的表现就是拒绝服务以及网络拥塞.

**消费组**

每个消费者都属于一个特定的消费组(通过group.id配置), 每个消费组都会选择一个broker作为消费组的协调器, 协调器来为每个消费者分配分区. 消费组中的消费者并发获取消息, 但是为了保证分区消息的有序性, 每个分区只会被一个消费者消费. 因此消费组中的消费者数量需要小于等于主题的分区个数.（如需全局消息有序, 只能有一个分区, 一个消费者）

一个主题可以对应多个消费组. 同一主题的一条消息只能被同一个消费组内的一个消费者消费, 但多个消费组也可以同时消费这一消息, 这是Kafka用来实现一个主题消息的广播（发给所有的Consumer）和单播（发给某一个Consumer）的手段. 如果需要实现广播, 只要每个消费者有一个独立的消费组就可以了. 要实现单播只需要所有的消费者在同一个消费组里.

**什么是分区重平衡?**

消费组里的一个消费者对应一个或多个分区, 当消费者加入或离开消费组时, 主题的分区就要进行重新分配, 这种现象称为重平衡（rebalance）. 以下情况会发生重平衡:

*  新的消费者加入消费组
 *  当前消费者从消费组退出（不管是异常退出还是正常关闭）
 *  消费者取消对某个主题的订阅
 *  订阅主题的分区增加(Kafka的分区数可以动态增加但是不能减少)
 *  broker宕机新的组协调器当选(因为是组协调器进行分区分配)
 *  当消费者在$\{session.timeout.ms\}时间内还没有发送心跳请求, 组协调器认为消费者已退出.

分区重平衡操作提供了消费者的高可用和高可扩展性, 但需要注意的是, 在重平衡过程中, 由于需要给消费者重新分配分区, 所以会出现在一个短暂时间内消费者不能拉取消息的状况, 这就导致消费性能下降, 所以我们需要考虑如何安全的进行重平衡以及如何尽可能避免重平衡.

> 注意一种慢消费者的情况, 当组协调器没有在session.timeout.ms时间内收到心跳请求或消费者没有在max.poll.interval.ms时间内消费完数据, 就会认为该消费者是慢消费者, 从而将其从消费组中移除, 触发重平衡操作. 在手动提交偏移量的模式下, 要特别注意这个问题, 否则会出现提交不了的情况, 导致一直在重复消费.

### Kafka集群 ###

一个Kafka服务器也称为broker, 若干个broker组成一个集群（cluster）, 其中集群内某个broker会成为集群控制器（cluster controller）, 它负责管理集群, 包括分配分区到broker、监控broker故障, 选举分区的主副本等.

**集群控制器的选举算法**

集群中的第一个broker通过在Zookeeper的/controller路径下创建一个临时节点来成为控制器, 当其他broker启动时, 也会试图创建一个临时节点, 但是会收到“节点已存在”的异常, 这样便知道集群控制器已存在. 这些broker会监听Zookeeper的这个控制器临时节点, 当控制器发生故障时, 该临时节点会消失, 这些broker便会收到通知, 然后尝试去创建临时节点成为新的控制器.

### 高可用机制 ###

Kafka的高可用机制主要体现在数据备份和主副本选举上.

**数据备份**

在Kafka中会为每个分区配置副本数(default.replication.factor), 每个分区可以有多个副本, 但只会有一个主副本(leader replica) ,其他都为从副本(follower replica), 所有的读写操作都发生在主副本上, 从副本只起一个冷备作用(即只起备份作用, 不提供读写).

> 注意这里的副本概念容易和其他中间件弄混, 其他中间件的副本配置需要手动开启, 而Kafka的副本配置是默认开启的(默认为1), 也就是说Kafka的一个分区就是一个主副本.  
> 举例说明, 其他中间件: 一个分区, n个副本. Kafka: 一个主副本, n-1个从副本.

(1) 为什么Kafka的从副本不提供读写, 只做备份?

因为如果follower副本也提供写服务的话, 那么就需要在所有的副本之间相互同步. n个副本就需要 n x n 条通路来同步数据, 如果采用异步同步的话, 数据的一致性和有序性是很难保证的, 而采用同步方式进行数据同步的话, 那么写入延迟其实是放大n倍的, 反而适得其反.

(2) 那么为什么不让follower副本提供读服务, 减少leader副本的读压力呢？

这个除了因为同步延迟带来的数据不一致之外, 不同于其他的存储服务（如ES，MySQL）, Kafka的读取本质上是一个有序的消息消费, 消费进度是依赖于一个叫做offset的偏移量, 这个偏移量是要保存起来的. 如果多个副本进行读负载均衡, 那么这个偏移量就不好确定了.

**主副本选举**

当集群中的一个broker离开时, 集群控制器会检查该broker是否有分区的主副本, 如果有, 控制器会在分区的ISR中选出一个新的主副本(ISR中的第一个, 不行就依次类推 ), 并通知新的主副本和其他的从副本. 如果控制器发现有新的broker（这个broker也有可能是之前宕机的）加入时, 它会通过此broker的ID来检查该broker是否有分区副本存在, 如果有则通知该broker从相应的分区主副本同步消息.

(1) 什么是ISR(in-sync replicas)?

ISR(in-sync replicas), 顾名思义, 就是跟leader副本 “保持同步” 的follower副本列表. 如果一段时间内(replica.lag.time.max.ms) leader副本没有收到follower副本的拉取请求, 或者这段时间内, follower副本没有fetch到leader副本的log end offset, 就会被leader副本从ISR中移除.

(2) 主副本同步消息的过程是怎样的呢?

生产者发布消息到某个分区时, 先通过ZooKeeper找到该分区的leader副本, 然后将消息只发送给leader副本, leader副本收到消息后将其写入本地磁盘. 接着每个follower副本都从leader副本上pull消息, follower副本收到消息后会向leader副本发送ACK(acknowledge). 一旦leader副本收到了 ISR (in-sync replicas) 中的所有副本的ACK, 该消息就被认为已经commit了, leader副本将增加 HW( High-Watermark) 并且向生产者发送ACK. 消费者读消息只会从leader副本中读取, 只有被commit过的消息（offset低于HW的消息）才会暴露给消费者.

> 为了提高性能, 每个follower副本在接收到消息后就立马向leader副本发送ACK, 而非等到消息写入磁盘中. 因此, 对于已经commit的消息, Kafka只能保证它被存于多个副本的内存中, 而不能保证它们被持久化到磁盘中, 也就不能完全保证异常发生后该条消息一定能被消费者消费.

(3) 什么是HW( High-Watermark)?

HW( High-Watermark), 这个offset是所有ISR的Log End Offset(最新同步完成消息的偏移量) , 消费者不能读取超过HW的消息, 因为这意味着读取到未完全同步的消息, 所以HW是消费者所能获取到的消息的最大offset.

### Kafka为什么速度那么快? ###

我们都知道Kafka的核心特性之一就是高吞吐率, 但Kafka的数据是存储在磁盘上的, 一般认为在磁盘上读写数据性能很低, 那Kafka是如何做到高吞吐率的呢?

*  批量压缩
 *  顺序写
 *  零拷贝

Kafka高吞吐率的秘诀在于, 它把所有的消息都进行批量压缩, 提升网络IO, 通过顺序写和零拷贝技术提升磁盘IO.

> 顺序写和零拷贝技术详细介绍: https://blog.csdn.net/litianxiang\_kaola/article/details/103632124

### Kafka的一些问题 ###

(1) 只保证单个主题单个分区内的消息有序, 但是不能保证单个主题所有分区消息有序.

(2) 消费偏移量由消费者跟踪和提交, 可能会存在消息被消费了, 但偏移量却没有提交成功, 导致消息重复消费的问题.

(3) 主题和分区的数目有限, 达到1000个主题左右时, 性能就开始下降.