[大数据] Kafka 介绍之订阅模式

在发布-订阅消息系统中，消息被持久化到一个topic中。与点对点消息系统不同的是，消费者可以订阅一个或多个topic，消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。在发布-订阅消息系统中，消息的生产者称为发布者，消费者称为订阅者。该模式的示例图如下：

发布者发送到topic的消息，只有订阅了topic的订阅者才会收到消息。

1.3 Kafka中的术语解释

???????broker： Kafka 集群包含一个或多个服务器，服务器节点称为broker。

? ? ? ? ? Broker是kafka实例，每个服务器上有一个或多个kafka的实例, 姑且认为每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号，如图中的broker-0、broker-1等……

???????Topic：?类似于数据库的表名

? ? ? ? 消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。

???????Partition?：?类似于数据库的行数据

? ? ? ?Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的，partition的表现形式就是一个一个的文件夹！

????????Replication:副本的作用是做备胎

? ? ? ?当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。

???????Producer：?生产者即数据的发布者

? ? ? ?该角色将消息发布到Kafka的topic中。broker接收到生产者发送的消息后，broker将该消息追加到当前用于追加数据的segment文件中。生产者发送的消息，存储到一个partition中，生产者也可以指定数据存储的partition。

???????Consumer：?消费者即数据的订阅者

????????消费者可以从broker中读取数据。消费者可以消费多个topic中的数据。

???????Consumer Group：消费者组

????????多个消费组组成一个消费者组，在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！。

???????Leader?：partition中负责数据的读写? ? ??

????????每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的partition。

???????Follower：?与Leader保持数据同步

????????Follower跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower，Follower与Leader保持数据同步。如果Leader失效，则从Follower中选举出一个新的Leader。当Follower与Leader挂掉、卡住或者同步太慢，leader会把这个follower从“in sync replicas”（ISR）列表中删除，重新创建一个Follower。

???????Zookeeper：

????????kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性。

? ? ? ?Message：每一条发送的消息主体。

1.4?Producer在写入数据

?发送的流程就在图中已经说明了，就不单独在文字列出来了！需要注意的一点是，消息写入leader后，follower是主动的去leader进行同步的！producer采用push模式将数据发布到broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的！写入示意图如下：

分区的主要目的是：

?　　1、方便扩展。因为一个topic可以有多个partition，所以我们可以通过扩展机器去轻松的应对日益增长的数据量。
　　2、提高并发。以partition为读写单位，可以多个消费者同时消费数据，提高了消息的处理效率。

那在kafka中，如果某个topic有多个partition，producer又怎么知道该将数据发往哪个partition呢？

???????1、 partition在写入的时候可以指定需要写入的partition，如果有指定，则写入。
　　2、如果没有指定partition，但是设置了数据的key，则会根据key的值hash出一个partition。
　　3、如果既没指定partition，又没有设置key，则会轮询选出一个partition。?

1.5? 保存数据?????????

　　Producer将数据写入kafka后，集群就需要对数据进行保存了！kafka将数据保存在磁盘，可能在我们的一般的认知里，写入磁盘是比较耗时的操作，不适合这种高并发的组件。Kafka初始会单独开辟一块磁盘空间，顺序写入数据（效率比随机写入高）。

????????Partition 结构
　　前面说过了每个topic都可以分为一个或多个partition，如果你觉得topic比较抽象，那partition就是比较具体的东西了！Partition在服务器上的表现形式就是一个一个的文件夹，每个partition的文件夹下面会有多组segment文件，每组segment文件又包含.index文件、.log文件、.timeindex文件（早期版本中没有）三个文件， log文件就实际是存储message的地方，而index和timeindex文件为索引文件，用于检索消息。

?　如上图，这个partition有三组segment文件，每个log文件的大小是一样的，但是存储的message数量是不一定相等的（每条的message大小不一致）。文件的命名是以该segment最小offset来命名的，如000.index存储offset为0~368795的消息，kafka就是利用分段+索引的方式来解决查找效率的问题。

????????Message结构
上面说到log文件就实际是存储message的地方，我们在producer往kafka写入的也是一条一条的message，那存储在log中的message是什么样子的呢？消息主要包含消息体、消息大小、offset、压缩类型……等等！我们重点需要知道的是下面三个：
　　1、 offset：offset是一个占8byte的有序id号，它可以唯一确定每条消息在parition内的位置！
　　2、消息大小：消息大小占用4byte，用于描述消息的大小。
　　3、消息体：消息体存放的是实际的消息数据（被压缩过），占用的空间根据具体的消息而不一样。

????????存储策略
　　无论消息是否被消费，kafka都会保存所有的消息。那对于旧数据有什么删除策略呢？
　　1、基于时间，默认配置是168小时（7天）。
　　2、基于大小，默认配置是1073741824。
　　需要注意的是，kafka读取特定消息的时间复杂度是O(1)，所以这里删除过期的文件并不会提高kafka的性能！

????????消费数据

　　消息存储在log文件后，消费者就可以进行消费了。与生产消息相同的是，消费者在拉取消息的时候也是找leader去拉取。

　　多个消费者可以组成一个消费者组（consumer group），每个消费者组都有一个组id！同一个消费组者的消费者可以消费同一topic下不同分区的数据，但是不会组内多个消费者消费同一分区的数据！！！

?图示是消费者组内的消费者小于partition数量的情况，所以会出现某个消费者消费多个partition数据的情况，消费的速度也就不及只处理一个partition的消费者的处理速度！如果是消费者组的消费者多于partition的数量，那会不会出现多个消费者消费同一个partition的数据呢？上面已经提到过不会出现这种情况！多出来的消费者不消费任何partition的数据。所以在实际的应用中，建议消费者组的consumer的数量与partition的数量一致！

假如现在需要查找一个offset为368801的message是什么样的过程呢？

建立在offset为有序的基础上，利用segment+有序offset+稀疏索引+二分查找+顺序查找等多种手段来高效的查找数据！至此，消费者就能拿到需要处理的数据进行处理了

????????1.6 kafka集群partition分布原理分析

在Kafka集群中，每个Broker都有均等分配Partition的Leader机会。
上述图Broker Partition中，箭头指向为副本，以Partition-0为例:broker1中parition-0为Leader，Broker2中Partition-0为副本。
上述图种每个Broker(按照BrokerId有序)依次分配主Partition,下一个Broker为副本，如此循环迭代分配，多副本都遵循此规则。
副本分配算法如下：
将所有N Broker和待分配的i个Partition排序.
将第i个Partition分配到第(i mod n)个Broker上.
将第i个Partition的第j个副本分配到第((i + j) mod n)个Broker上.

????????1.7 Zookeeper在kafka的作用

无论是kafka集群，还是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些meta信息。
Kafka使用zookeeper作为其分布式协调框架，很好的将消息生产、消息存储、消息消费的过程结合在一起。
同时借助zookeeper，kafka能够生产者、消费者和broker在内的所以组件在无状态的情况下，建立起生产者和消费者的订阅关系，并实现生产者与消费者的负载均衡。

????????1.8 Kafka的特性

高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作；
可扩展性：kafka集群支持热扩展；
持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失；
容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）；
高并发：支持数千个客户端同时读写；
支持实时在线处理和离线处理：可以使用Storm这种实时流处理系统对消息进行实时进行处理，同时还可以使用Hadoop这种批处理系统进行离线处理；

参考：

1?https://www.cnblogs.com/sujing/p/10960832.html

2??https://www.cnblogs.com/diaozhaojian/p/10571542.html

3 kafka?工作流程深入解析