进击消息中间件系列（八）：Kafka 主题与分区

民工哥

发布于 2023-08-22 14:16:14

3730

发布于 2023-08-22 14:16:14

主题和分区是Kafka的两个核心概念，主题作为消息的归类，可以再细分为一个或者多个分区，分区可以看作是对消息的二次归类。分区的划分不仅可以为Kafka提供了可伸缩性，水平扩展能力，还可以通过副本机制来为Kafka提供数据冗余以提高数据的可靠性，为了做到均匀分布，通常partition的数量通常是BrokerServer数量的整数倍。

从底层上来说，主题和分区都是逻辑上的概念。分区可以有一个或多个副本，每个副本对应一个日志文件，每个日志文件对应一个或多个日志段文件。

主题、分区、副本、日志关系：

主题的管理

创建主题

bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic topicone --partitions 2 --replicationfactor 1

localhost:2181 #zookeeper所在的ip，zookeeper 必传参数，多个zookeeper用 ‘，’分开。
partitions #用于设置主题分区数，每个线程处理一个分区数据
replication-factor #用于设置主题副本数，每个副本分布在不通节点，不能超过总结点数。如你只有一个 节点，但是创建时指定副本数为2，就会报错。

查看主题

查看topic元数据

topic元数据保存在Zookeeper节点中 (zookeeper命令 )

[zk: localhost:2181(CONNECTED) 0] get /brokers/topics/topicone
{"version":1,"partitions":{"1":[0],"0":[0]}}
cZxid = 0x33
ctime = Sat Nov 06 21:04:57 PDT 2021
mZxid = 0x33
mtime = Sat Nov 06 21:04:57 PDT 2021
pZxid = 0x35
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 44
numChildren = 1

查看所有主题

执行kafka命令

[root@localhost zookeeper-3.4.14]# cd /opt/kafka/kafka_2.12-2.2.1/
[root@localhost kafka_2.12-2.2.1]# bin/kafka-topics.sh --list -zookeeper localhost:2181
__consumer_offsets
topicone
[root@localhost kafka_2.12-2.2.1]#

查看特定主题

[root@localhost kafka_2.12-2.2.1]# bin/kafka-topics.sh --describe -zookeeper localhost:2181 --topic topicone
Topic:topicone PartitionCount:2 ReplicationFactor:1 Configs:
 Topic: topicone Partition: 0 Leader: 0 Replicas: 0 Isr: 0
 Topic: topicone Partition: 1 Leader: 0 Replicas: 0 Isr: 0
[root@localhost kafka_2.12-2.2.1]#

修改主题

// 增加配置 
bin/kafka-topics.sh --alter -zookeeper localhost:2181 --topic topicone --config flush.messages=1

Default(默认值)：None
server.properties：log.flush.interval.messages
说明(解释)： 
log文件”sync”到磁盘之前累积的消息条数,因为磁盘IO操作是一个慢操作,但又是一个”数据可靠性"的必要手段,所以此参数的设置,需要在"数据可靠性"与"性能"之间做必要的权衡.如果此值过大,将会导致每次"fsync"的时间较长(IO阻塞),如果此值过小,将会导致"fsync"的次数较多,这也意味着整体的client请求有一定的延迟.物理server故障,将会导致没有fsync的消息丢失.

删除主题

若 delete.topic.enable=true 直接彻底删除该 Topic。

若 delete.topic.enable=false 如果当前 Topic 没有使用过即没有传输过信息：可以彻底删除。

如果当前 Topic 有使用过即有过传输过信息：并没有真正删除 Topic 只是把这个 Topic 标记为删除(marked for deletion)，重启 Kafka Server 后删除。

bin/kafka-topics.sh --delete -zookeeper localhost:2181 --topic topicone
topicone is marked for deletion.

增加分区

// 增加分区数
[root@localhost kafka_2.12-2.2.1]# bin/kafka-topics.sh --alter -zookeeper localhost:2181 --topic topicone --partitions 3
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!
[root@localhost kafka_2.12-2.2.1]#

修改分区数时，仅能增加分区个数。若是用其减少 partition 个数，会报错。

KafkaAdminClient应用

我们都习惯使用Kafka中bin目录下的脚本工具来管理查看Kafka，但是有些时候需要将某些管理查看的功能集成到系统（比如Kafka Manager）中，那么就需要调用一些API来直接操作Kafka了。

package com.demo.kafkademo.ch4;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;
/**
 * KafkaAdminClient应用
 */
public class KafkaAdminConfigOperation {
    static String brokerList =  "192.168.33.129:9092";
    static String topic = "topicone";
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        describeTopicConfig();
        //alterTopicConfig();
        //addTopicPartitions();
    }
    // 查看主题详情
    public static void describeTopicConfig() throws ExecutionException,
            InterruptedException {

        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        AdminClient client = AdminClient.create(props);

        ConfigResource resource =
                new ConfigResource(ConfigResource.Type.TOPIC, topic);
        DescribeConfigsResult result =
                client.describeConfigs(Collections.singleton(resource));
        Config config = result.all().get().get(resource);
        System.out.println("=====================================");
        System.out.println("config:"+config);
        client.close();
    }
 // 修改主题配置
    public static void alterTopicConfig() throws ExecutionException, InterruptedException {

        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        AdminClient client = AdminClient.create(props);

        ConfigResource resource =
                new ConfigResource(ConfigResource.Type.TOPIC, topic);
        ConfigEntry entry = new ConfigEntry("cleanup.policy", "compact");
        Config config = new Config(Collections.singleton(entry));
        Map<ConfigResource, Config> configs = new HashMap<>();
        configs.put(resource, config);
        AlterConfigsResult result = client.alterConfigs(configs);
        result.all().get();

        client.close();
    }
 // 添加分区
    public static void addTopicPartitions() throws ExecutionException, InterruptedException {

        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        AdminClient client = AdminClient.create(props);

        NewPartitions newPartitions = NewPartitions.increaseTo(5);
        Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
        newPartitionsMap.put(topic, newPartitions);
        CreatePartitionsResult result = client.createPartitions(newPartitionsMap);
        result.all().get();

        client.close();
    }
}

分区

主要内容 : kafka分区的管理，分区重新分配。

Kafka可以将主题划分为多个分区（Partition），会根据分区规则选择把消息存储到哪个分区中，只要如果分区规则设置的合理，那么所有的消息将会被均匀的分布到不同的分区中，这样就实现了负载均衡和水平扩展。另外，多个订阅者可以从一个或者多个分区中同时消费数据，以支撑海量数据处理能力。

由于消息是以追加到分区中的，多个分区顺序写磁盘的总效率要比随机写内存还要高（引用Apache Kafka – A High Throughput Distributed Messaging System的观点），是Kafka高吞吐率的重要保证之一。

副本机制

由于Producer和Consumer都只会与Leader角色的分区副本相连，所以kafka需要以集群的组织形式提供主题下的消息高可用。kafka支持主备复制，所以消息具备高可用和持久性。

一个分区可以有多个副本，这些副本保存在不同的broker上。每个分区的副本中都会有一个作为 Leader。当一个broker失败时，Leader在这台broker上的分区都会变得不可用，kafka会自动移除 Leader，再其他副本中选一个作为新的Leader。

在通常情况下，增加分区可以提供kafka集群的吞吐量。然而，也应该意识到集群的总分区数或是单台服务器上的分区数过多，会增加不可用及延迟的风险。

// topicone 主题有三个分区,一个副本同处在一个节点中
[root@localhost kafka_2.12-2.2.1]# bin/kafka-topics.sh --describe -zookeeper localhost:2181 --topic topicone
Topic:topicone PartitionCount:3 ReplicationFactor:1 Configs:
 Topic: topicone Partition: 0 Leader: 0 Replicas: 0 Isr: 0
 Topic: topicone Partition: 1 Leader: 0 Replicas: 0 Isr: 0
 Topic: topicone Partition: 2 Leader: 0 Replicas: 0 Isr: 0
[root@localhost kafka_2.12-2.2.1]#

分区Leader选举

可以预见的是,如果某个分区的Leader挂了,那么其它跟随者将会进行选举产生一个新的leader,之后所有的读写就会转移到这个新的Leader上。

在kafka中,其不是采用常见的多数选举的方式进行副本的Leader 选举,而是会在Zookeeper上针对每个Topic维护一个称为ISR（in-sync replica，已同步的副本）的集合, 显然还有一些副本没有来得及同步。只有这个ISR列表里面的才有资格成为leader(先使用ISR里面的第一个，如果不行依次类推，因为ISR里面的是同步副本，消息是最完整且各个节点都是一样的)。

通过 ISR,kafka需要的冗余度较低，可以容忍的失败数比较高。假设某个topic有f+1个副本，kafka可以容忍f 个不可用,当然,如果全部ISR里面的副本都不可用,也可以选择其他可用的副本,只是存在数据的不一致。

分区重新分配

我们往已经部署好的Kafka集群里面添加机器是最正常不过的需求，而且添加起来非常地方便，我们需要做的事是从已经部署好的Kafka节点中复制相应的配置文件，然后把里面的broker id修改成全局唯一的，最后启动这个节点即可将它加入到现有Kafka集群中。

创建集群和主题

首先创建一个有三个节点的集群。创建主题 (topic-cluster) 3个分区 , 副本数为 3。

主题的副本数。每个主题可以有多个副本，副本位于集群中不同的broker上，也就是说副本的数量不能超过broker的数量，否则创建主题时会失败。

[root@localhost opt]# cd kafka/kafka-server-01
[root@localhost kafka-server-01]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic topic-cluster --partitions 3 --replication-factor 3
Created topic topic-cluster.
[root@localhost kafka-server-01]#

查看主题详情

[root@localhost kafka-server-01]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic-cluster
Topic:topic-cluster PartitionCount:3 ReplicationFactor:3 Configs:
 Topic: topic-cluster Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
 Topic: topic-cluster Partition: 1 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
 Topic: topic-cluster Partition: 2 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2
[root@localhost kafka-server-01]#

分析 :

可以看到 topic-cluster 这个主题 , PartitionCount:3 有三个分区;ReplicationFactor:3 (代表三个副本)
Topic: topic-cluster Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
0分区: leader在broker 1 上, 副本平均分部在 broker 0 ,1, 2 上

再添加一个分区

[root@localhost kafka-server-01]# bin/kafka-topics.sh --alter --zookeeper localhost:2181 --topic topic-cluster --partitions 4
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!
[root@localhost kafka-server-01]#

再次查看主题详情:

[root@localhost kafka-server-01]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic-cluster
Topic:topic-cluster PartitionCount:4 ReplicationFactor:3 Configs:
 Topic: topic-cluster Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
 Topic: topic-cluster Partition: 1 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
 Topic: topic-cluster Partition: 2 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2
 Topic: topic-cluster Partition: 3 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
[root@localhost kafka-server-01]#

可以看到在3个节点的集群上又增加一个分区后, 分区0 和分区3 多在 1节点上 , 这样就导致了分配不均衡, 节点1的压力增大。

再添加一个节点

复制kafka -> kafka-server04 主要修改server.properties : 节点id , 日志路径, 端口号。

[root@localhost kafka]# cp -r kafka_2.12-2.2.1 kafka-server-04
[root@localhost kafka]# ll
total 55720
drwxr-xr-x. 7 root root      101 Nov  6 20:48 kafka_2.12-2.2.1
-rw-r--r--. 1 root root 57054917 Jun 19  2019 kafka_2.12-2.2.1.tgz
drwxr-xr-x. 7 root root      101 Nov 17 21:14 kafka-server-01
drwxr-xr-x. 7 root root      101 Nov 17 21:14 kafka-server-02
drwxr-xr-x. 7 root root      101 Nov 17 21:14 kafka-server-03
drwxr-xr-x. 7 root root      101 Nov 17 23:22 kafka-server-04

查看主题信息

[root@localhost kafka-server-04]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic-cluster
Topic:topic-cluster PartitionCount:4 ReplicationFactor:3 Configs:
 Topic: topic-cluster Partition: 0 Leader: 1 Replicas: 1,2,0Isr: 1,2,0
 Topic: topic-cluster Partition: 1 Leader: 2 Replicas: 2,0,1Isr: 2,0,1
 Topic: topic-cluster Partition: 2 Leader: 0 Replicas: 0,1,2Isr: 0,1,2
 Topic: topic-cluster Partition: 3 Leader: 1 Replicas: 1,2,0Isr: 1,2,0
[root@localhost kafka-server-04]#

重新分配

现在需要将原先分布在broker 1-3节点上的分区重新分布到broker 1-4节点上，借助kafkareassign-partitions.sh工具生成reassign plan，不过先得按照要求定义一个文件，里面说明哪些 topic需要重新分区，文件内容如下：

{"topics":[{"topic":"topic-cluster"}], "version":1 }

然后使用kafka-reassign-partitions.sh工具生成reassign plan

[root@localhost kafka-server-01]# bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file reassign.json --broker-list "0,1,2,3" --generate

–topics-to-move-json-file #指定分区重分配对应的主题的分配明细路径

注: 如果复制因子比broker 代理个数大 , 则会报错:

Partitions reassignment failed due to Replication factor: 4 larger than available brokers: 1.

分区重分配失败，原因是复制因子:4比可用代理大:1。

执行结果:

Current partition replica assignment
{"version":1,"partitions":[{"topic":"topic-cluster","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"topic-cluster","partition":1,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"topic-cluster","partition":0,"replicas":[1,2,0],"log_dirs":["any","any","any"]},{"topic":"topic-cluster","partition":3,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"topic-cluster","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"topic-cluster","partition":3,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"topic-cluster","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"topic-cluster","partition":2,"replicas":[2,3,0],"log_dirs":["any","any","any"]}]}

命令输出了两个Json字符串:

第一个JSON内容为当前的分区副本分配情况，
第二个为重新分配的候选方案，注意这里只是生成一份可行性的方案，并没有真正执行重分配的动作。

将第二个JSON内容保存到名为assignplan.json文件里面（文件名不重要，文件格式也不一定要以json为结尾，只要保证内容是json即可），然后执行 reassign plan：

格式化一下即将重新分配的明细json。

{
 "version": 1,
 "partitions": [
  {"topic": "topic-cluster","partition": 1,"replicas": [1,2,3],"log_dirs": ["any","any","any"]},
  {"topic": "topic-cluster","partition": 3,"replicas": [3,0,1],"log_dirs": ["any","any","any"]},
  {"topic": "topic-cluster","partition": 0,"replicas": [0,1,2],"log_dirs": ["any","any","any"]},
  {"topic": "topic-cluster","partition": 2,"replicas": [2,3,0],"log_dirs": ["any","any","any"]}
 ]
}

保存后执行分区重新分配

[root@localhost kafka-server-01]# bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file assignplan.json --execute
Current partition replica assignment

{"version":1,"partitions":[{"topic":"topic-cluster","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"topic-cluster","partition":1,"replicas":[2,0,1],"log_dirs":["any","any","any"]},{"topic":"topic-cluster","partition":0,"replicas":[1,2,0],"log_dirs":["any","any","any"]},{"topic":"topic-cluster","partition":3,"replicas":[1,2,0],"log_dirs":["any","any","any"]}]}

Save this to use as the --reassignment-json-file option during rollback
Successfully started reassignment of partitions.
[root@localhost kafka-server-01]#

查看分配进度

[root@localhost kafka-server-01]# bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file assignplan.json -verify Status of partition reassignment:
Status of partition reassignment: 
Reassignment of partition topic-cluster-1 completed successfully
Reassignment of partition topic-cluster-3 completed successfully
Reassignment of partition topic-cluster-0 completed successfully
Reassignment of partition topic-cluster-2 completed successfully
[root@localhost kafka-server-01]#

全部完成

查看topic-cluster

[root@localhost kafka-server-01]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic-cluster
Topic:topic-cluster PartitionCount:4 ReplicationFactor:3 Configs:
 Topic: topic-cluster Partition: 0 Leader: 1 Replicas: 0,1,2 Isr: 1,2,0
 Topic: topic-cluster Partition: 1 Leader: 1 Replicas: 1,2,3 Isr: 2,1,3
 Topic: topic-cluster Partition: 2 Leader: 2 Replicas: 2,3,0 Isr: 0,2,3
 Topic: topic-cluster Partition: 3 Leader: 3 Replicas: 3,0,1 Isr: 1,0,3

修改副本因子

实际项目中我们可能在创建topic时没有设置好正确的replication-factor，导致kafka集群虽然是高可用的，但是该topic在有broker宕机时，可能发生无法使用的情况。topic一旦使用又不能轻易删除重建，因此动态增加副本因子就成为最终的选择。

说明：kafka 1.0版本配置文件默认没有default.replication.factor=x，因此如果创建topic时，不指定– replication-factor 想，默认副本因子为1. 我们可以在自己的server.properties中配置上常用的副本因子，省去手动调整。例如设置default.replication.factor=3。

配置topic的副本，保存为json文件。

{
"version":1, "partitions":[
            {"topic":"topic-cluster2","partition":0,"replicas":[0,1,2]},
      {"topic":"topic-cluster2","partition":1,"replicas":[0,1,2]},
      {"topic":"topic-cluster2","partition":2,"replicas":[0,1,2]}
        ] 
}

然后执行脚本

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file replication-factor.json --execute

分区策略

按照Kafka默认的消费逻辑设定，一个分区只能被同一个消费组（ConsumerGroup）内的一个消费者消费。假设目前某消费组内只有一个消费者C0，订阅了一个topic，这个topic包含7个分区，也就是说这个消费者C0订阅了7个分区，参考下图。

此时消费组内又加入了一个新的消费者C1，按照既定的逻辑需要将原来消费者C0的部分分区分配给消费者C1消费，情形上图（2），消费者C0和C1各自负责消费所分配到的分区，相互之间并无实质性的干扰。

接着消费组内又加入了一个新的消费者C2，如此消费者C0、C1和C2按照上图（3）中的方式各自负责消费所分配到的分区。

如果消费者过多，出现了消费者的数量大于分区的数量的情况，就会有消费者分配不到任何分区。参考下图，一共有8个消费者，7个分区，那么最后的消费者C7由于分配不到任何分区进而就无法消费任何消息。

上面各个示例中的整套逻辑是按照Kafka中默认的分区分配策略来实施的。Kafka提供了消费者客户端参数partition.assignment.strategy用来设置消费者与订阅主题之间的分区分配策略。默认情况下，此参数的值为：org.apache.kafka.clients.consumer.RangeAssignor，即采用RangeAssignor分配策略。

除此之外，Kafka中还提供了另外两种分配策略：RoundRobinAssignor和StickyAssignor。消费者客户端参数partition.asssignment.strategy可以配置多个分配策略，彼此之间以逗号分隔。

RangeAssignor分配策略

参考源码：org.apache.kafka.clients.consumer.RangeAssignor

RangeAssignor策略的原理是按照消费者总数和分区总数进行整除运算来获得一个跨度，然后将分区按照跨度进行平均分配，以保证分区尽可能均匀地分配给所有的消费者。对于每一个topic， RangeAssignor策略会将消费组内所有订阅这个topic的消费者按照名称的字典序排序，然后为每个消费者划分固定的分区范围，如果不够平均分配，那么字典序靠前的消费者会被多分配一个分区。

假设n=分区数/消费者数量，m=分区数%消费者数量，那么前m个消费者每个分配n+1个分区，后面的（消费者数量-m）个消费者每个分配n个分区。

假设消费组内有2个消费者C0和C1，都订阅了主题t0和t1，并且每个主题都有4个分区，那么所订阅的所有分区可以标识为：t0p0、t0p1、t0p2、t0p3、t1p0、t1p1、t1p2、t1p3。最终的分配结果为：

消费者C0：t0p0、t0p1、t1p0、t1p1 
消费者C1：t0p2、t0p3、t1p2、t1p3

假设上面例子中2个主题都只有3个分区，那么所订阅的所有分区可以标识为：t0p0、t0p1、t0p2、 t1p0、t1p1、t1p2。最终的分配结果为：

消费者C0：t0p0、t0p1、t1p0、t1p1 
消费者C1：t0p2、t1p2

可以明显的看到这样的分配并不均匀，如果将类似的情形扩大，有可能会出现部分消费者过载的情况。

RoundRobinAssignor分配策略

参考源码：org.apache.kafka.clients.consumer.RoundRobinAssignor

RoundRobinAssignor策略的原理是将消费组内所有消费者以及消费者所订阅的所有topic的partition按照字典序排序，然后通过轮询方式逐个将分区以此分配给每个消费者。RoundRobinAssignor策略对应的partition.assignment.strategy参数值为：org.apache.kafka.clients.consumer.RoundRobinAssignor。

假设消费组中有2个消费者C0和C1，都订阅了主题t0和t1，并且每个主题都有3个分区，那么所订阅的所有分区可以标识为：t0p0、t0p1、t0p2、t1p0、t1p1、t1p2。最终的分配结果为：

消费者C0：t0p0、t0p2、t1p1 
消费者C1：t0p1、t1p0、t1p2

如果同一个消费组内的消费者所订阅的信息是不相同的，那么在执行分区分配的时候就不是完全的轮询分配，有可能会导致分区分配的不均匀。如果某个消费者没有订阅消费组内的某个topic，那么在分配分区的时候此消费者将分配不到这个topic的任何分区。

假设消费组内有3个消费者C0、C1和C2，它们共订阅了3个主题：t0、t1、t2，这3个主题分别有1、2、 3个分区，即整个消费组订阅了t0p0、t1p0、t1p1、t2p0、t2p1、t2p2这6个分区。具体而言，消费者 C0订阅的是主题t0，消费者C1订阅的是主题t0和t1，消费者C2订阅的是主题t0、t1和t2，那么最终的分配结果为：

消费者C0：t0p0 
消费者C1：t1p0 
消费者C2：t1p1、t2p0、t2p1、t2p2

可以看到RoundRobinAssignor策略也不是十分完美，这样分配其实并不是最优解，因为完全可以将分区t1p1分配给消费者C1。

StickyAssignor

分配策略参考源码：org.apache.kafka.clients.consumer.StickyAssignor Kafka从0.11.x版本开始引入这种分配策略，它主要有两个目的：

分区的分配要尽可能的均匀；
分区的分配尽可能的与上次分配的保持相同。

当两者发生冲突时，第一个目标优先于第二个目标。鉴于这两个目标，StickyAssignor策略的具体实现要比RangeAssignor和RoundRobinAssignor这两种分配策略要复杂很多。

假设消费组内有3个消费者：C0、C1和C2，它们都订阅了4个主题：t0、t1、t2、t3，并且每个主题有2 个分区，也就是说整个消费组订阅了t0p0、t0p1、t1p0、t1p1、t2p0、t2p1、t3p0、t3p1这8个分区。最终的分配结果如下：

消费者C0：t0p0、t1p1、t3p0 
消费者C1：t0p1、t2p0、t3p1 
消费者C2：t1p0、t2p1

假设此时消费者C1脱离了消费组，那么消费组就会执行再平衡操作，进而消费分区会重新分配。如果采用RoundRobinAssignor策略，那么此时的分配结果如下：

消费者C0：t0p0、t1p0、t2p0、t3p0 
消费者C2：t0p1、t1p1、t2p1、t3p1

如分配结果所示，RoundRobinAssignor策略会按照消费者C0和C2进行重新轮询分配。而如果此时使用的是StickyAssignor策略，那么分配结果为：

消费者C0：t0p0、t1p1、t3p0、t2p0 
消费者C2：t1p0、t2p1、t0p1、t3p1

可以看到分配结果中保留了上一次分配中对于消费者C0和C2的所有分配结果，并将原来消费者C1的“负担”分配给了剩余的两个消费者C0和C2，最终C0和C2的分配还保持了均衡。

自定义分配策略

需实现：org.apache.kafka.clients.consumer.internals.PartitionAssignor 继承自：org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor

参考文章:https://nopabuleng.blog.csdn.net/article/ details/121412322 https://nopabuleng.blog.csdn.net /article/details/121412360

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2023-08-02，如有侵权请联系 cloudcommunity@tencent.com 删除

kafka

本文分享自民工哥技术之路微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度