分布式消息队列 -- kafka基础知识相关

客官°小女子只卖身不卖艺 2023-10-05 08:12 129阅读 0赞

**Kafkas是什么？**

**Kafka**是一个分布式消息队列, 也是一个开源消息系统**，**由**Scala**语言写成**。****kafka**对消息保存时根据**Topic**进行归类**，**发送消息者称为**Producer，**消息接受者称为**Consumer，**此外**kafka**集群有多个**kafka**实例组成**，**每个实例(**server**)称为**broker。**无论是**kafka**集群**，**还是**consumer**都依赖于**zookeeper**集群保存一些**mate**信息**，**来保证系统可用性。在通常的流式计算中**，****kafka** 一般用来缓存数据**，**而**Storm**或者**Spark Streaming**通过消费**kafka**的数据进行计算**。**

**Kafka架构分析图：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70][]

**(1)Producer:**消息生产者，就是向**kafka broker**发消息的客户端。  
**(2)Consumer:**消息消费者，向**kafka broker**取消息的客户端。  
**(3)Topic:**可以理解为一个队列。

**(4)ConsumerGroup(CG):**这是**kafka**用来实现一个**topic**消息的广播（发给所有的**consumer**）和单播（发给任意一个**consumer**）的手段。一个**topic**可以有多个**CG**。**topic** 的消息会复制（不是真的复制**，**是概念上的）到所有的**CG，**但每个**partition**只会把消息发给该**CG**中的一个**consumer**。如果需要实现广播，只要每个**consumer**有一个独立的**CG**就可以了。要实现单播只要所有的**consumer**在同一个**CG**。用**CG**还可以将**consumer**进行自由的分组而不需要多次发送消息到不同的**topic。**  
**(5)Broker**：一台**kafka**服务器就是一个**broker。**一个集群由多个**broker**组成。一个**broker**可以容纳多个**topic**。  
**(6)Partition:**为了实现扩展性，一个非常大的**topic**可以分布到多个**broker**（即服务器）上**，**一个**topic**可以分为多个**partition**，每个**partition**是一个有序的队列。**partition**中的每条消息都会被分配一个有序的**id**（**offset**）。**kafka**只保证按一个**partition**中的顺序将消息发给**consumer**，不保证一个**topic**的整体（多个**partition** 间）的顺序。  
**(7)Offset: kafka**的存储文件都是按照**offset.kafka**来命名**，**用**offset**做名字的好处是方便查找。例如你想找位于**2049**的位置，只要找到**2048.kafka**的文件即可**。**当然**the first offset**就是**00000000000.kafka**

#### **Kafka生产过程分析：** ####

**(1)**写入方式**:****producer**采用推(**push**)模式将消息发布到**broker**，每条消息都被追加 (**append**)到分区**(partition**)中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障**kafka**吞吐率）。

**(2)**分区(**Partition):**消息发送时都被发送到一个**topic**，其本质就是一个目录，而**topic**是由一些**Partition Logs**(分区日志)组成， 其组织结构如下图所示**：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70 2][]

每个**Partition**中的消息都是有序的**，**生产的消息被不断追加到**Partition log**上**，**其中的每一个消息都被赋予了一个唯一的**offset**值**。**

**(3)**分区的原因：  
**a****、**方便在集群中扩展**，**每个**Partition**可以通过调整以适应它所在的机器，而一个**topic**又可以有多个**Partition**组成**，**因此整个集群就可以适应任意大小的数据了。  
b、可以提高并发**，**因为可以以**Partition**为单位读写了。

**(4)**分区的原则：

**a****、**指定了**partition****，**则直接使用。

1.  未指定**partition**但指定**key****，**通过对**key**的**value**进行**hash**出一个**partition**。
2.  **partition**和**key**都未指定**，**使用轮询选出一个**partition**。

**(5)**副本(**Replication)**:

同一个**partition**可能会有多个**replication**(对应**server.properties**配置中的**default.replication.factor=N**）。没有**replication**的情况下，一旦**broker**宕机，其上所有**patition**的数据都不可被消费，同时**producer** 也不能再将数据存于其上的**patition**。引入**replication**之后，同一个**partition**可能会有多个**replication**，而这时需要在这些**replication**之间选出一个**leader**，**producer**和**consumer**只与这个**leader** 交互，其它**replication**作为**follower**从**leader**中复制数据。

**(6)**写入流程：

**producer**写入消息流程如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70 3][]

**a****、****producer**先从**zookeeper**的**"/brokers/.../state"**节点找到该**partition**的**leader****。**  
**b****、****producer**将消息发送给该**leader****。**  
**c****、****leader**将消息写入本地**log****。**  
**d****、****followers**从**leader pull**消息**，**写入本地**log**后向**leader**发送**ACK****。**

**e****、****leader**收到所有**ISR**中的**replication**的**ACK**后**，**增加**HW****（****high watermark****，**最后**commit**的**offset****）**并向**producer**发送**ACK****。**

**(7)Broker**保存消息

**a****、**存储方式

物理上把**topic**分成一个或多个**partition**（对应**server.properties**中的**num.partitions=3**配置），每个**partition**物理上对应一个文件夹（该文件夹存储该 **partition** 的所有消息和索引文件）。

**b****、**存储策略

无论消息是否被消费，**kafka**都会保留所有消息。有两种策略可以删除旧数据基于时间: **log.retention.hours=168**

基于大小: **log.retention.bytes=1073741824**  
需要注意的是，因为**kafka**读取特定消息的时间复杂度为**O(1)**，即与文件大小无关，所以删除过期文件与提高**kafka**性能无关。

#### **Kafka生产过程分析：** ####

**kafka**提供了两套**consumer API**：高级**Consumer API**和低级**Consumer API**。

**(1)**高级API

**优点**：

高级**API**写起来简单。  
不需要自行去管理**offset**，系统通过**zookeeper**自行管理。  
不需要管理分区，副本等情况，系统自动管理。  
消费者断线会自动根据上一次记录在**zookeeper**中的**offset**去接着获取数据（默认设置**1**分钟更新一下**zookeeper**中存的**offset****）**可以使用**group**来区分对同一个**topic**的不同程序访问分离开来（不同的**group**记录不同的**offset**，这样不同程序读取同一个**topic**才不会因为**offset**互相影响）。

**缺点**：

不能自行控制**offset****（**对于某些特殊需求来说**）**。  
不能细化控制如分区、副本、**zk** 等。

**(2)**低级**API**

**优点**：

能够让开发者自己控制**offset**，想从哪里读取就从哪里读取。  
自行控制连接分区，对分区自定义进行负载均衡。  
对**zookeeper**的依赖性降低（如：**offset** 不一定非要靠**zk**存储，自行存储**offset**即可，比如存在文件或者内存中）。

**缺点**：

太过复杂，需要自行控制**offset**，连接哪个分区，找到分区**leader** 等。

**(3)****消费者组**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70 4][]

消费者是以**consumer group** 消费者组的方式工作，由一个或者多个消费者组成一个组，共同消费一个**topic**。每个分区在同一时间只能由**group**中的一个消费者读取**，**但是多个**group**可以同时消费这个**partition**。在图中，有一个由三个消费者组成的**group****，**有一个消费者读取主题中的两个分区，另外两个分别读取一个分区。某个消费者读取某个分区，也可以叫做某个消费者是某个分区的拥有者。  
在这种情况下**，**消费者可以通过水平扩展的方式同时读取大量的消息。另外，如果一个消费者失败了，那么其他的**group** 成员会自动负载均衡读取之前失败的消费者读取的分区。

**(4)****消费方式**

**consumer**采用**pull**（拉）模式从**broker**中读取数据**。****push**（推）模式很难适应消费速率不同的消费者**，**因为消息发送速率是由**broker**决定的**。**它的目标是尽可能以最快速度传递消息**，**但是这样很容易造成**consumer**来不及处理消息**，**典型的表现就是拒绝服务以及网络拥塞**。**而**pull**模式则可以根据**consumer**的消费能力以适当的速率消费消息**。**对于**kafka**而言，**pull**模式更合适**，**它可简化**broker**的设计**，****consumer**可自主控制消费消息的速率**，**同时**consumer**可以自己控制消费方式—即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义**。****pull**模式不足之处是，如果**kafka**没有数据**，**消费者可能会陷入循环中**，**一直等待数据到达**。**为了避免这种情况，可以在拉请求中有参数**，**允许消费者请求在等待数据到达的**“**长轮询**”**中进行阻塞（并且可选地等待到给定的字节数，以确保大的传输大小）**。**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200813102142640.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200813102408639.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200813102423200.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200813102502325.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200813103925370.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NIWUxPR08=,size_16,color_FFFFFF,t_70