Kafka 分布式消息系统基础概念剖析

迷南。 2022-12-04 02:18 160阅读 0赞

**目录**

生产者 & 消费者

Kafka 实例 broker

分区（Partition）

副本（Replication）

消费者组（Consumer Group）

消息发送流程

消息存储方式

消息过期策略

Zookeeper 存储结构

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

理论结合实践才能加深理解，本文是理论，实践可以参考：[Kafka 之 CentOS 上安装部署、集群启动、命令行操作][Kafka _ CentOS]

# 生产者 & 消费者 #

1、生产者负责产生消息和发送消息到消息系统，而消息系统将为消费者提供消息用于处理。

2、Kafka 是基于**发布/订阅**模式的消息系统，生产者会将消息推送（push）到 Kafka 中的某个 Topic（主题）上，消费者根据需要订阅相应的 Topic 获取消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70][]

3、consumer 采用 pull（拉）模式从 broker 中读取数据。push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由 broker 决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成 consumer 来不及处理消息，典型的表现就是拒绝服务以及网络拥塞，而 pull 模式则可以根据 consumer 的消费能力以适当的速率消费消息。即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。

# Kafka 实例 broker #

1、一个 Broker 就是一个 Kafka 实例，一台机器上可运行一个或多个 Broker，多个 Broker 就组成了一个 Kafka 集群。

2、Topic（主题） 可以指定副本数量，作为多个**副本（**replication**）**位于多台机器上，每个 Broker 上都有一份相同的 Topic。

3、Kafka 使用 [ZooKeeper ][ZooKeeper]在多个 Broker 中选举出一个 Leader（领导者），其他副本将作为 Follower（追随者），Leaer 主要负责读写消息，也就是和生产者、消费者打交道，同时将消息同步写到其他副本中。

4、当有 Broker 失效时，如果 Topic 没有了 Leader，则会重新选举出新的 Leader，继续提供服务。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 1][]

#  分区（Partition） #

1、Kafka 将 Topic（主题） 拆分成多个**分区**（partitions），类似数据库的分库，这样起到了负载均衡的作用，可不受单机的限制。

<table> 
 <tbody> 
  <tr> 
   <td> <p><span style="color:#f33b45;">创建 Topic 命令：</span>./kafka-topics.sh --create --bootstrap-server 192.168.116.128:9092 --<span style="color:#f33b45;">replication-factor </span>2 --<span style="color:#f33b45;">partitions </span>2 --<span style="color:#f33b45;">topic </span>helloWorld</p> <p>--<span style="color:#f33b45;">replication-factor </span>：用于指定副本个数、--<span style="color:#f33b45;">partitions </span>：用于指定分区个数、--<span style="color:#f33b45;">topic </span>：指定主题名称</p> </td> 
  </tr> 
 </tbody> 
</table>

2、分区（partitions）的本质就是 kafka 服务器下的一个目录，设置了几个分区就有几个目录，消息会发送到指定的 topic（主题）上，不同分区存储的消息也是不同的！。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 2][]

3、每个 Partition（分区） 中的消息都是有序的，生产的消息被不断追加（appended）到 Partition log 上，其中的每一个消息都被赋予了一个唯一的 offset（偏移量） 值。

# 副本（Replication） #

1、同一个 partition（分区） 通常会有多个 replication（对应 server.properties 中的 default.replication.factor=1 配置），副本的实质就是在集群的各个 kafka 中节点中复制相同的 partition（分区）文件。

2、没有 replication（副本） 的情况下，一旦 broker 宕机，其上所有 patition 的数据都不可被消费，同时 producer 也不能再将数据存于其上的 patition；引入 replication 之后，同一个 partition 可能会有多个 replication，此时即使某个 broker 宕机，其余的仍然可以正常使用。

# **消费者组（Consumer Group）** #

1、为了进一步提高吞吐量，Kafka 引入了**消费者组**的概念，由一个或者多个消费者组成一个组，相当于是消费者的集群。

2、比如消费者 A 是一个应用 A 实例，为了提高消费的吞吐量，可以多部署了几个消费者 A 实例（多实例），这样就有多个消费者形成一个消费组，但干的都是应用 A 做的事，需要与消费者B（不同的应用）区分开。

3、一般设置消费组的消费者数量与分区数（partitions）一致，这样一个消费者能负责一个分区，提高效率。如果消费组的消费者数量小于分区数，则会出现一个消费者负责多个分区。而如果消费组的消费者数量大于分区数，则会出现有消费者分不到分区，造成浪费。所以一般保持一致。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 3][]

4、消费者组共同消费一个 topic，每个分区在同一时间只能由 group 中的一个消费者读取，但是多个 group 可以同时消费这个partition。某个消费者读取某个分区，也可以叫做某个消费者是某个分区的拥有者。

5、在这种情况下，消费者可以通过水平扩展的方式同时读取大量的消息，另外，如果一个消费者失败或者宕机了，那么 group 中的其他成员会自动负载均衡的读取它的分区数据/消息。

# 消息发送流程 #

1、producer（生产者） 采用推送（push）模式将消息发布到 broker（kafka节点），每条消息都被追加（append）到分区（patition）中，顺序写入磁盘（顺序写磁盘效率比随机写内存要高，保障 kafka 吞吐率）。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 4][]

<table> 
 <tbody> 
  <tr> 
   <td>1）producer（生产者）先从 zookeeper 的 "/brokers/.../state" 节点找到该集群中的 leader（领导者）</td> 
  </tr> 
  <tr> 
   <td>2）producer 将消息发送给该 leader</td> 
  </tr> 
  <tr> 
   <td>3）leader 将消息写入本地 log（kafka 以本地文件存储数据）</td> 
  </tr> 
  <tr> 
   <td>4）followers（追随者） 从 leader（领导者） pull（拉取） 消息，写入本地 log 后向 leader 发送 ACK（确认收到消息）</td> 
  </tr> 
  <tr> 
   <td> <p style="margin-left:0cm;">5）leader 收到集群中所有 replication 的 ACK 后，向 producer 发送 ACK（确认消息发送成功）</p> </td> 
  </tr> 
 </tbody> 
</table>

2、一个分区可以看做是一个单独的队列，producer （生产者）根据策略将消息写入对应的分区，策略有三种：

<table> 
 <tbody> 
  <tr> 
   <td> <p>一、直接指定分区；</p> <p>二、如果未指定分区，则根据消息的 key，通过哈希函数计算得出分区；</p> <p>三、如果没有 key，则轮询选出一个分区。</p> <p><span style="color:#f33b45;">注意：分区中的数据是不同的，一条消息只会进入一个分区，而消费者组中的消费者则会根据偏移量去分区中取得相应的消息进行消费处理。</span></p> </td> 
  </tr> 
 </tbody> 
</table>

# 消息存储方式 #

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

4 drwxr-xr-x. 2 root root  4096 9月   7 11:27 helloWorld-0
     4 drwxr-xr-x. 2 root root  4096 9月   7 11:27 helloWorld-1
    [root@wangmaoxiong001 logs]# cd helloWorld-0/
    [root@wangmaoxiong001 helloWorld-0]# ll
    总用量 4
    -rw-r--r--. 1 root root 10485760 9月   7 11:27 00000000000000000000.index
    -rw-r--r--. 1 root root        0 9月   7 11:27 00000000000000000000.log
    -rw-r--r--. 1 root root 10485756 9月   7 11:27 00000000000000000000.timeindex
    -rw-r--r--. 1 root root        8 9月   7 11:27 leader-epoch-checkpoint
    [root@wangmaoxiong001 helloWorld-0]#

# 消息过期策略 #

1、无论消息是否被消费，kafka 都会保留所有消息，有两种策略可以删除旧数据，可以通过 server.properties 文件配置过期策略（两种过期策略是独立的）：

<table> 
 <tbody> 
  <tr> 
   <td>log.retention.hours=168：基于时间方式删除过期数据，超时将被删除<span style="color:#f33b45;">，默认 168 小时（7 天）</span></td> 
  </tr> 
  <tr> 
   <td>log.retention.bytes=1073741824：基于大小的日志（数据）保留策略，当存储的数据量超过此大小时，就删除旧数据。<span style="color:#f33b45;">默认为 1G。</span>此配置默认是被注释的。</td> 
  </tr> 
 </tbody> 
</table>

2、特别提醒：因为 Kafka 读取特定消息的时间复杂度为 O(1)，与文件大小无关，所以删除过期文件与提高 Kafka 性能无关。

# Zookeeper 存储结构 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 5][]

[Kafka _ CentOS]: https://wangmaoxiong.blog.csdn.net/article/details/107725519
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70]: /images/20221123/915225b591b94b7489ca821bae45755d.png
[ZooKeeper]: https://wangmaoxiong.blog.csdn.net/article/details/107732098#Zookeeper%20%E9%80%89%E4%B8%BE%E6%9C%BA%E5%88%B6
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 1]: /images/20221123/d9380f1d63a44d5e97eae8e766e49430.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 2]: /images/20221123/831ae1eba18b4feabe331d3adc962224.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 3]: /images/20221123/caac228d33994bb0b437422c493435bc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 4]: /images/20221123/8934ef6765b64b95bd8508c38c36106a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70 5]: /images/20221123/20c28560cc364c9e87288ab84b6c0b58.png