Kafka 分区机制详解

一、前言

由于负责产品的性质原因，我需要大量接触 Kafka，因此对 Kafka 的使用和原理都有一定的了解！

这一期来聊聊 Kafka 非常非常重要的分区机制：

• 主题与分区之间的关系 ✅
• 分区工作的原理流程 ✅
• 如何创建一个多分区的主题 ✅

二、主题与分区

在第一期的时候聊过，Kafka 是基于发布-订阅模型而构建，生产者向主题发送消息，而消费者则通过订阅主题来消费消息。

而主题里面又可以创建多个分区，新建的主题默认只有一个分区。

我们来看看 Kafka 的基础架构：

一个 Kafka 集群内有若干个「实例」，当我们创建一个「主题」时，会默认在「主题」内创建一个「分区」，我们的消息就是发往这个「分区」中。

但一个「主题」内可不可以创建多个「分区」呢？

当然可以，我们可以通过修改配置文件去控制「主题」创建时创建的「分区数」，也可以在创建「主题」的时候，指定「分区数」。多个「分区」可以分布在不同的「实例」中，但「分区」本身是不可分割的，同一个「分区」不可能出现在两个「实例」中。

到这里就很清楚了，一个主题内可以有多个分区，而不同的分区可以分布在不同的实例，生产端和消费端可以指定主题，也可以指定主题中的若干个分区来工作。

四、分区的多副本机制

Kafka 为了提高消息存储的安全性和容灾能力，引入了多副本机制。

多副本机制是针对 Partition 设计的，可以这么理解，Kafka 为 Partition 拷贝了多份，并且这个份数是可以指定的，多份 Partition 中属于主从关系，只有一个 Leader，其他叫 Follower，当 Leader 不行了，挂了，会从所有Follower中挑一个合格的上位，与 Leader 同步程度达不到要求的都是不够资格的小弟，不参加 Leader 选举。

我们平时其实只跟 Leader 打交道，发送的消息都在 Leader 中，消息提交到 Leader 后其他 Follower 会自动从 Leader 中拉取同步。

1. Leader 选举原理

我们先来了解一个概念：ISR（In-sync Replicas）同步的副本集合，在这个集合内的副本被认为是完全同步的，包括 Leader 也存在于这个集合中。

Broker 有一个参数叫 replica.lag.time.max.ms，用于控制 Follower 能落后 Leader 多少时间，落后不超过这个时间的副本 Kafka 就认为它是同步的，落后超过这个时间该 Follower 会从 ISR 中踢出。

若是 Leader 挂了，集合也空了，这个分区就不可用了，因此，生产环境中，合理的配置副本也是非常重要的一项。

2. 选举规则

1. ISR 不为空，直接从 ISR 中选举
2. ISR 为空，Kafka 也可以从不在 ISR 中的存活副本中选举，这个过程称为 Unclean 领导者选举

对于第二个选项，可以通过 Broker 端的 unclean.leader.election.enable 参数来控制。

开启 Unclean 领导者选举可能会造成数据丢失，但好处是，它使得分区 Leader 副本一直存在，不至于停止对外提供服务，因此提升了高可用性，不推荐，毕竟可用性可以通过其他手段来提升。

反之，禁止 Unclean 领导者选举的好处在于维护了数据的一致性，避免了消息丢失，但牺牲了高可用性。

三、分区的工作流程

生产者分区器根据参数来区分数据要发往的分区：

1. 有 Partition：直接将数据存入对应的分区
2. 没有 Partition 有 Key：将通过 Key的Hash值 % 主题的分区数 来得到一个 Partition 值
3. 没有 Partition 没有 Key：采用 Sticky Partition（粘性分区器），会随机选择一个分区，并尽可能的一直使用这个分区，待该分区的 ProducerBatch 满了或者已完成，再随机选择其他的分区（不会重复使用上一次的分区）

消费者可以用两种方式消费分区里面的数据：

1. 指定主题，不指定分区：该主题下的所有分区的数据，订阅该主题的消费者都能消费到，但需要注意的是，不同分区之间的数据是不能保证消费顺序的
2. 指定主题，指定分区：可以指定任意分区进行消费

五、实战：创建一个多分区的主题

1. 使用命令行创建

以 Windows 脚本为例，从官网获取安装包，跟需要操作的 Kafka 版本对上即可。

使用 Kafka 安装包中自带的命令行脚本创建主题：

# 创建一个名为「test1」 的主题，其拥有五个分区，每个分区拥有一个副本
.\kafka-topics.bat --bootstrap-server localhost:9092 --create --topic test1 --partitions 5 --replication-factor 1

# 查看主题「test1」的详情
.\kafka-topics.bat --bootstrap-server localhost:9092 --describe --topic test1

Topic: test1    PartitionCount: 5       ReplicationFactor: 1    Configs: segment.bytes=1073741824,max.message.bytes=157286400,delete.retention.ms=1800000
        Topic: test1    Partition: 0    Leader: 0       Replicas: 0     Isr: 0
        Topic: test1    Partition: 1    Leader: 0       Replicas: 0     Isr: 0
        Topic: test1    Partition: 2    Leader: 0       Replicas: 0     Isr: 0
        Topic: test1    Partition: 3    Leader: 0       Replicas: 0     Isr: 0
        Topic: test1    Partition: 4    Leader: 0       Replicas: 0     Isr: 0

2. 使用 API 创建

引入依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.2.0</version>
</dependency>

使用 AdminClient 创建一个名为「test2」 的主题，其拥有五个分区，每个分区拥有一个副本：

import java.util.Collections;
import java.util.Properties;
public class Test {
    public static void main(String[] args)  {
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "10.0.0.110:9092");
        // 创建 AdminClient 对象
        AdminClient adminClient = AdminClient.create(properties);
        adminClient.createTopics(Collections.singletonList(new NewTopic("test2", 5, (short) 1)));
        // 关闭资源
        adminClient.close();
    }
}

查看主题详情：

.\kafka-topics.bat --bootstrap-server 10.0.0.110:9092 --describe --topic test2

Topic: test2    PartitionCount: 5       ReplicationFactor: 1    Configs: segment.bytes=1073741824,max.message.bytes=157286400,delete.retention.ms=1800000
        Topic: test2    Partition: 0    Leader: 0       Replicas: 0     Isr: 0
        Topic: test2    Partition: 1    Leader: 0       Replicas: 0     Isr: 0
        Topic: test2    Partition: 2    Leader: 0       Replicas: 0     Isr: 0
        Topic: test2    Partition: 3    Leader: 0       Replicas: 0     Isr: 0
        Topic: test2    Partition: 4    Leader: 0       Replicas: 0     Isr: 0