常见消息队列总结-Kafka

水深无声 2022-05-27 04:52 262阅读 0赞

Kafka的设计初衷是希望作为一个统一的信息收集平台，能够实时的收集反馈信息，并需要能够支撑较大的数据量，且具备良好的容错性。主要设计目标如下：

*  以时间复杂度O(1)的方式提供消息持久化能力，并保证即使对TB级以上数据也能保证常数时间的访问性能。
 *  高吞吐率，即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。
 *  支持Kafka Server间的消息分区，及分布式消费消息，同时保证每个partition内的消息顺序传输。
 *  同时支持离线数据处理和实时数据处理。

## 概念说明 ##

*  Broker：Kafka集群包含一个或多个服务器，这种服务器称为Broker。Kafka支持水平扩展，一般Broker数量越多，集群吞吐率越高。
 *  Topic： 每条发布到Kafka集群的消息都由一个类别，这个类别被成为topic  
    
    
     *  物理上不同topic的消息分开存储，逻辑上一个topic的消息虽然保存于一个或多个broker上但用户只需要指定消息的topic即可生产或消费数据。
 *  Partition：Partition是物理上的概念，创建topic时可以指定partition的数量，每个topic包含一个或多个Partition。
 *  Producer： 负责发布消息到Kafka broker
 *  Consumer：消费消息，每个Consumer属于一个特定的Consumer Group，同一topic的一条消息只能被同一Consumer Group内的一个Consumer消费，**但多个Consumer Group可同时消费这一消息。**

## 架构说明 ##

![这里写图片描述][70]

![这里写图片描述][70 1]

如上图所示，一个典型的Kafka集群中包含若干个Producer（可以是Web前端产生的Page View，或者是服务器日志，系统CPU，Memory等），若干个Broker（Kafka支持水平扩展，一般Broker数量越多，集群吞吐率越高），若干个Consumer Group，以及一个Zookeeper集群，Kafka通过Zookeeper管理集群配置，选举Leader，以及在Consumer Group发生变化是进行Rebalance。Producer使用Push模式将消息将消息发布到Broker，Consumer使用pull模式从broker订阅并消费消息。

## 特性说明 ##

*  Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer，接收消息的称为Consumer，Kafka集群由多个Kafka(broker)实例组成。Kafka集群几乎不需要维护任何Consumer和Producer状态信息，这些信息由Zookeeper保存。因此Producer和Consumer的客户端实现非常轻量级，它们可以随意离开，不会对集群造成影响。
 *  一个Topic可认为是一类消息，每个Topic被分为多个Partition，每个Partition在存储层面append log文件。每条消息在文件中的位置称为offse（偏移量），Kafka没有提供其他额外的索引机制来存储offset，所以Kafka几乎不允许对消息进行**随机读写** （Consumer在消费消息时，将offset重置为任意值，便可实现随机读写，但实际情况该操作总是无意义的）。
 *  消息被消费后不会立即被删除，日志文件将会根据broker中的配置要求保留一段时间后再删除。原因是**Kafka通过这种简单的手段，来释放磁盘空间，以减少消息消费后对文件内容改动的磁盘IO开销**
 *  Partition设计的**目的** ，Kafka基于文件存储，通过分区，可以将日志内容分散到多个broker上，**避免文件尺寸达到单机磁盘的上限** ，每个Partition都会被当前broker保存，可以将一个topic切分为任意多个Partition，来提高（消息保存/消息消费）速率，更多的Partition，能够有效提升并发消费的能力。
 *  Kafka可以配置Partition需要备份的个数（replicas），每个Partition将会被备份到多台机器上，以提高可用性。
 *  Producer能够决定消息归属于那个Partition，比如基于“round robin”方式或者其他的一些算法。
 *  **本质上Kafka只支持Topic** ，每个Consumer属于一个Consumer Group，每个Group中可以有多个Consumer，发送到Topic的消息，只会被订阅此Topic的每个Group中的一个Consumer消费。  
    
    
     *  如果所有的Consumer都具有相同的Group，这种模式类似于queue模式，消息会在Consumers之间负载均衡。
     *  如果所有的Consumer都具有不同的Group，那这就是“发布-订阅”，消息将会广播给所有的消费者。

## 使用场景 ##

*  Messaging：partitions/replication和容错，是kafka具有良好的扩展性和性能优势，所以对于一些常规的消息系统是不错的选择。但Kafka并没有提供**事务性** 、**消息确认机制** 、**消息分组** 等企业级特性，所以Kafka并未确保消息的发送与接收绝对可靠（比如消息重复、消息发送丢失等）。
 *  Websit Activity Tracking：Kafka可以作为**网站活性跟踪** 的最佳工具，可以将网页/用户操作等信息发送到Kafka中，可以进行实时监控或者离线分析统计等。
 *  Log Aggregation：Kafka的特性决定它非常适合作为**日志收集中心** ，replication可以将操作日志**批量** 、**异步** 的发送到Kafka集群中，而不是保存在本地或者DB中，Kafka可以批量提交消息/压缩消息等，这对producer端而言感觉不到性能的开支，此时consumer端可以使Hadoop等其他系统化的存储和分析系统。

## 设计原理 ##

### 持久性 ###

Kafka使用文件存储消息，决定Kafka在性能上严重依赖文件系统本身特性。Kafka是对日志文件进行append操作，所以**磁盘检索** 的开销较小，同时为了减少磁盘I/O调用次数，broker会将消息暂时buffer起来，当消息的个数达到阈值时，才会flush到磁盘。

### 性能 ###

影响到Kafka的吞吐量的除了磁盘IO外，还有网络IO。对于Producer端，采用将消息buffer起来，批量处理的方式；对于Consumer端，采用批量fetch多条消息，消息量的大小可以通过配置文件来指定；对于broker端，可以使用Sendfile系统调用，将文件的数据映射到系统内存中，无需进程再进行Copy和交互。对于上述的三者改进来说，CPU的开销应该都不大，所以启用**消息压缩机制** (Kfaka支持gzip、snappy等多种压缩方式）是一个良好的策略。压缩需要消耗少量的CPU资源，不过对于提升Kfaka的网络IO来说，该操作是值得的。

### 生产者 ###

Producer将会和Topic下所有的Partition Leader保持Socket连接，消息将直接通过Socket发送到broker，中间不会经过任何“路由层”。

*  Partition Leader的位置（host:port）注册在zookeeper中，producer作为zookeeper client，已经注册了watch用来监听Partition Leader变更事件。
 *  将多条消息暂存在客户端buffer起来，批量延迟发送提升了网络效率，带来的**隐患**是，当Producer失效时，尚未发送的消息将会丢失。

### 消费者 ###

Topic的模型中Consumer消费消息有pull/push两种方式。不过在Kafka中，采用了pull方式，即Consumer在和broker建立连接之后，主动去pull（或者说fetch）消息，该模式的优点是Consumer端可以根据自己的消费能力适时的去fetch消息并处理，且可以控制消息消费的进度（offset）。

*  其他常见的消息系统中，消息消费的位置、控制消息的状态由Prodiver保留，以避免重复发送消息等。
 *  在Kafka中，Partition中的消息只有一个Consumer在消费，所以不存在状态控制，也没有复杂的消息确认机制，由此可推断**kafka broker端是轻量级的** 。Consumer接收消息后，本地保存最后消息的offset，并**间歇向Zookeeper注册offset** 。
 *  Push模式很难适应消费速率不同的Consumer，因为消费发送速率是由Broker决定的，Push模式的目标是尽可能以最快速度传递消息，但是这样很容易造成Consumer来不及处理消息，**造成消息堆积**。

### 消息传送机制 ###

*  at most once：最多一次，发送一次，无论成败，将不会重发。
    
    消费者fetch消息后先保存offset，然后处理消息，如果在消息处理过程中出现了异常，导致部分消息未能继续处理，那么**未处理的消息不能被fetch** ，这就是“at most once”。
 *  at least once：至少一次，如果消息未接收成功，可能会重发，直到接收成功。
    
    消费者fetch消息后先处理消息，然后保存offset，如果保存offset阶段，Zookeeper异常导致保存操作未能执行成功，下次fetch时可能获得上次已经处理过的消息，这就是“at least once”，**原因offset没有及时的提交给Zookeeper，Zookeeper仍记录之前的offset**
 *  exactly once： 只发送一次。
    
    kafka中没有严格实现exactly once（基于2阶段提交事务）。

### 复制备份 ###

Kafka提供将每个partition数据复制到多个broker上能力，任何一个partition有一个leader和零至多个follower。

*  备份的个数可以通过broker配置文件来设定。
 *  leader处理所有的read/write请求，并将请求同步给follower，leader负责跟踪所有的follower状态（**如果一个follower落后太多，可将其从replicas表中删除**） ，follower将leader同步的请求保存到本地。  
    
    
     *  当所有的follower都将该消息保存成功，此消息才被认为是**committed** ，只有处于committed状态的消息，consumer才能消费它。
     *  即使只有一个replicas实例存活，在zookeeper集群存活的状态，仍然可以保证消息的正常发送和接受。
 *  leader失效重选时，需要注意两点，一必须选择一个“up to date”的follower，二是新的leader对应的broker是否已经有多个partition leader，**过多的partition leader意味着该broker上将承受更多的IO压力** （需要考虑负载均衡）。

### 日志 ###

创建一个topic包含2个partition，那么日志将会保存在topic\_0和topic\_1两个目录中。日志文件保存了一序列“log entries”（日志条目），**每个log entry格式为”4个字节的数字N表示消息的长度“ + ”N个字节的消息内容** 。每个日志都一个offset来唯一标记一条消息，offset的值为8字节的数字，表示此消息在此Partition的起始位置。每个Partition在物理存储层面，**有多个log file组成（称为segment）** ，segmentfile的命名为**“最小offset”.Kafka** ，其中最小offset表示此segment中起始消息的offset。

![这里写图片描述][70 2]

*  每个Partition中所持有的segment列表信息会存储到zookeeper中。
 *  当segment文件尺寸达到一定阈值时（可以通过配置文件设定，默认是1G），将会创建一个新的文件。
 *  当buffer中消息的条数达到阈值将会触发日志信息flush到日志文件中，同时如果“距离最近一次flush的时间差” 达到阈值时，也会触发flush到日志文件。
 *  获取消息时，需要指定offset和最大chunk尺寸，offset用来表示消息的起始位置，chunk size用来表示最大获取消息的总长度（**间接的表示消息的条数** ）。
 *  日志文件的删除策略非常简单，启动一个后台线程定期扫描log file列表，把保存时间超过阈值的文件直接删除。

### 分配 ###

Kafka使用zookeeper来存储一些meta信息，并使用zookeeper watch机制来发现meta信息的变更并作出相应的动作（比如consumer失效，触发负载均衡等）。

*  Broker node registry：当一个Kafka broker启动后，首先会向Zookeeper注册自己的节点信息（临时znode），当broker和Zookeeper断开连接时，此znode也会被删除。
    
     *  格式：/broker/ids/\[0…N\] -> host: port 其中\[0…N\]表示broker id，每个broker的配置文件都需要指定一个数字类型的id（全局不可重复），znode的值为此broker的host: port信息。
 *  Broker Topic Registry： 当一个broker启动时，会向zookeeper注册自己持有的topic和partitions信息，仍然是一个临时znode。
    
     *  格式： /broker/topics/\[topic\]/\[0…N\]
 *  Consumer and Consumer group：每个consumer客户端被创建时,会向zookeeper注册自己的信息，此作用主要是为了”负载均衡”。
    
     *  一个group中的多个consumer可以交错的消费一个topic的所有partitions，简而言之，保证此topic的所有partitions都能被此group所消费，且消费时为了性能考虑，让partition相对均衡的分散到每个consumer上。
 *  Consumer id Registry：每个consumer都有一个唯一的ID(host:uuid,可以通过配置文件指定，也可以由系统生成)，此id用来标记消费者信息。
    
     *  格式：/consumers/\[group\_id\]/ids/\[consumer\_id\]
     *  仍然是一个临时的znode,此节点的值为\{“topic\_name”:\#streams…\},即表示此consumer目前所消费的topic + partitions列表.
 *  Consumer offset Tracking：用来跟踪每个consumer目前所消费的partition中**最大的offset**
    
     *  格式：/consumers/\[group\_id\]/offsets/\[topic\]/\[broker\_id-partition\_id\]–>offset\_value
     *  此znode为持久节点，可以看出offset跟group\_id有关，以表明**当group中一个消费者失效，其他consumer可以继续消费**
 *  Partition Owner registry： 用来标记partition被哪个consumer消费，临时znode。
    
     *  格式:/consumers/\[group\_id\]/owners/\[topic\]/\[broker\_id-partition\_id\]–>consumer\_node\_id当consumer启动时,所触发的操作:
     *  首先进行”Consumer id Registry”
        
         *  然后在”Consumer id Registry”节点下注册一个watch用来监听当前group中其他consumer的”leave”和”join”;只要此znode path下节点列表变更,都会触发此group下consumer的负载均衡.(比如一个consumer失效,那么其他consumer接管partitions).
         *  在”Broker id registry”节点下,注册一个watch用来监听broker的存活情况;如果broker列表变更,将会触发所有的groups下的consumer重新balance.

![这里写图片描述][70 3]

#### 总结 ####

*  Producer端使用zookeeper用来”发现”broker列表,以及和Topic下每个partition leader建立socket连接并发送消息。
 *  Broker端使用zookeeper用来注册broker信息,已经监测partitionleader存活性。
 *  Consumer端使用zookeeper用来注册consumer信息,其中包括consumer消费的partition列表等,同时也用来发现broker列表,并和partition leader建立socket连接,并获取消息。

## 参考资料 ##

*  [Kafka架构和原理深度剖析][Kafka]
 *  [Kafka入门：简介、使用场景、设计原理、主要配置及集群搭建][Kafka 1]
 *  [Kafka系列之(1)-Kafka概念和架构][Kafka_1_-Kafka]

[70]: /images/20220527/62c812acf2504223ae0fc4b7a5bb4aaf.png
[70 1]: /images/20220527/07432e21e2ea45abb951cc73236f6b7d.png
[70 2]: /images/20220527/1f1dacd366e8489fa026a1e81a0bb630.png
[70 3]: /images/20220527/fdcc92f717d94ad29f43ce9e48df2cbd.png
[Kafka]: http://lxw1234.com/archives/2015/09/504.htm
[Kafka 1]: http://www.cnblogs.com/likehua/p/3999538.html
[Kafka_1_-Kafka]: https://www.jianshu.com/p/b8cea98c17b3