Kafka_入门

客官°小女子只卖身不卖艺 2022-01-09 23:59 301阅读 0赞

# **介绍** #

**百度百科**

Kafka是一种高吞吐量的[分布式][Link 1]发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像[Hadoop][]一样的[日志][Link 2]数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过[Hadoop][]的并行加载机制来统一线上和离线的消息处理，也是为了通过[集群][Link 3]来提供实时的消息。

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

**Kafka专用术语：**

*  Broker：消息中间件处理结点，一个Kafka节点就是一个broker，多个broker可以组成一个Kafka集群。
 *  Topic：一类消息，Kafka集群能够同时负责多个topic的分发。
 *  Partition：topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。
 *  Segment：partition物理上由多个segment组成。
 *  offset：每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中。partition中的每个消息都有一个连续的序列号叫做offset，用于partition唯一标识一条消息
 *  partition:当存在多副本的情况下，会尽量把多个副本，分配到不同的broker上。

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70][]

大概用法就是，Producers往Brokers里面的指定Topic中写消息，Consumers从Brokers里面拉去指定Topic的消息，然后进行业务处理。  
图中有两个topic，topic 0有两个partition，topic 1有一个partition，三副本备份。可以看到consumer gourp 1中的consumer 2没有分到partition处理，这是有可能出现的，下面会讲到。

关于broker、topics、partitions的一些元信息用zk来存，监控和路由啥的也都会用到zk。

--------------------

## 生产过程 ##

创建一条记录，记录中一个要指定对应的topic和value，key和partition可选。 先序列化，然后按照topic和partition，放进对应的发送队列中。kafka produce都是批量请求，会积攒一批，然后一起发送，不是调send()就进行立刻进行网络发包。  
如果partition没填，那么情况会是这样的：

1.  key有填  
    按照key进行哈希，相同key去一个partition。（如果扩展了partition的数量那么就不能保证了）
2.  key没填  
    round-robin来选partition

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 1][]

**分区(Partition)**

Kafka集群有多个消息代理服务器（broker-server）组成，发布到Kafka集群的每条消息都有一个类别，用主题（topic）来表示。通常，不同应用产生不同类型的数据，可以设置不同的主题。一个主题一般会有多个消息的订阅者，当生产者发布消息到某个主题时，订阅了这个主题的消费者都可以接收到生产者写入的新消息。

Kafka集群为每个主题维护了分布式的分区（partition）日志文件，物理意义上可以把主题（topic）看作进行了分区的日志（partition log）。主题的每个分区都是一个有序的、不可变的记录序列，新的消息会不断追加到日志中，分区中的每条消息都会按照时间顺序分配到一个单调递增的顺序编号，叫做偏移量（offset），这个偏移量能够唯一地定位当前分区中的每一条消息

发布到Kafka主题的每条消息包括键值和时间戳，消息到达服务器端的指定分区后，都会分配到一个自增的偏移量，原始的消息内容和分配的偏移量以及其他一些元数据信息最后都会存储到分区日志文件中，消息的键也可以不用设置，这种情况下消息会均衡地分布到不同的分区

当存在多副本的情况下，会尽量把多个副本，分配到不同的broker上。**kafka会为partition选出一个leader，之后所有该partition的请求，实际操作的都是leader，然后再同步到其他的follower。**当一个broker歇菜后，所有leader在该broker上的partition都会重新选举，选出一个leader。（这里不像分布式文件存储系统那样会自动进行复制保持副本数）

生产流程

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 2][]

然后这里就涉及两个细节：怎么分配partition，怎么选leader。

**关于partition的分配，还有leader的选举，总得有个执行者。在kafka中，这个执行者就叫controller。**kafka使用zk在broker中选出一个controller，用于partition分配和leader选举。

**partition的分配**

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

**leader容灾**

controller会在Zookeeper的/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的机器活过来。

--------------------

## 消费 ##

订阅topic是以一个消费组来订阅的，一个消费组里面可以有多个消费者。同一个消费组中的两个消费者，不会同时消费一个partition。换句话来说，**就是一个partition，只能被消费组里的一个消费者消费**，但是可以同时被多个消费组消费。因此，如果消费组内的消费者如果比partition多的话，那么就会有个别消费者一直空闲。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 3][]

### 分配partition--reblance ###

生产过程中broker要分配partition，**消费过程这里，也要分配partition给消费者。类似broker中选了一个controller出来，消费也要从broker中选一个coordinator，用于分配partition。**

下面从顶向下，分别阐述一下

1.  怎么选coordinator。
2.  交互流程。
3.  reblance的流程。

### **选coordinator** ###

1.  看offset保存在那个partition
2.  该partition leader所在的broker就是被选定的coordinator

**offset的保存**

kafka把这个offset的保存，保存在一个名叫\_\_consumeroffsets topic的topic中。写进消息的key由groupid、topic、partition组成，value是偏移量offset。topic配置的清理策略是compact。总是保留最新的key，其余删掉。一般情况下，每个key的offset都是缓存在内存中，查询的时候不用遍历partition，如果没有缓存，第一次就会遍历partition建立缓存，然后查询返回。

**交互流程**

把coordinator选出来之后，就是要分配了  
整个流程是这样的：

*  consumer启动、或者coordinator宕机了，consumer会任意请求一个broker，发送ConsumerMetadataRequest请求，broker会按照上面说的方法，选出这个consumer对应coordinator的地址。
 *  consumer 发送heartbeat请求给coordinator，返回IllegalGeneration的话，就说明consumer的信息是旧的了，需要重新加入进来，进行reblance。返回成功，那么consumer就从上次分配的partition中继续执行。

转载url  [https://www.jianshu.com/p/d3e963ff8b70][https_www.jianshu.com_p_d3e963ff8b70]

[Link 1]: https://baike.baidu.com/item/%E5%88%86%E5%B8%83%E5%BC%8F/19276232
[Hadoop]: https://baike.baidu.com/item/Hadoop
[Link 2]: https://baike.baidu.com/item/%E6%97%A5%E5%BF%97/2769135
[Link 3]: https://baike.baidu.com/item/%E9%9B%86%E7%BE%A4/5486962
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70]: /images/20220109/73c0b4f3a2704abc9623c1200689dc8a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 1]: /images/20220109/0d30cb9f9d574545ad9f2d54d7014198.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 2]: /images/20220109/f1d1ec84648e434398f0b2cacd52199c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 3]: /images/20220109/dc4be04971aa41e0aa8803919b265f94.png
[https_www.jianshu.com_p_d3e963ff8b70]: https://www.jianshu.com/p/d3e963ff8b70