杰霸霸博客

消费者组中分区的分配以及再平衡

  1. 一个consumer group中有多个consumer组成，一个Topic有多个Partition组成，现在的问题是，到底由哪个consumer来消费哪个Partition的数据。
  2. Kafka有四种主流的分区分配策略： Range、RoundRobin、Sticky、CooperativeSticky。可以通过配置参数partition.assignment.strategy，修改分区的分配策略。默认策略是Range + CooperativeSticky。Kafka可以同时使用多个分区分配策略。

1. 消费者组中分区的分配

Alt text 参数说明：

参数名称 描述
heartbeat.interval.ms Kafka消费者和coordinator之间的心跳时间，默认3s。该条目的值必须小于session.timeout.ms，也不应该高于session.timeout.ms的1/3。
session.timeout.ms Kafka消费者和coordinator之间连接超时时间，默认45s。超过该值，该消费者被移除，消费者组执行再平衡。
max.poll.interval.ms 消费者处理消息的最大时长，默认是5分钟。超过该值，该消费者被移除，消费者组执行再平衡。
partition.assignment.strategy 消费者分区分配策略 , 默认策略是Range + CooperativeSticky。Kafka可以同时使用多个分区分配策略。可以选择 的策略包括： Range 、RoundRobin 、Sticky 、CooperativeSticky

2. Range策略以及再平衡

2.1 Range分区策略原理

Alt text

2.2 Range分区代码实操

  1. 修改testtopic分区，改成7个
sh
[jack@hadoop107 kafka-3.6.1]$ ./bin/kafka-topics.sh --bootstrap-server 192.168.101.107:9092 --alter --topic testtopic --partitions 7
[jack@hadoop107 kafka-3.6.1]$ ./bin/kafka-topics.sh --bootstrap-server 192.168.101.107:9092 --describe --topic testtopic
Topic: testtopic	TopicId: MnxtaR5OSxGVvnLX2c8KJQ	PartitionCount: 7	ReplicationFactor: 3	Configs: 
	Topic: testtopic	Partition: 0	Leader: 1	Replicas: 1,2,3	Isr: 3,1,2
	Topic: testtopic	Partition: 1	Leader: 2	Replicas: 2,3,1	Isr: 3,1,2
	Topic: testtopic	Partition: 2	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2
	Topic: testtopic	Partition: 3	Leader: 4	Replicas: 4,1,2	Isr: 4,1,2
	Topic: testtopic	Partition: 4	Leader: 1	Replicas: 1,3,4	Isr: 1,3,4
	Topic: testtopic	Partition: 5	Leader: 2	Replicas: 2,4,1	Isr: 2,4,1
	Topic: testtopic	Partition: 6	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2
  1. 同时启动3个消费者, 指定消费者组test666, 指定消费主题为testtopic。
java
public static void main(String[] args) {

    Properties properties = new Properties();

    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.101.105:9092");

    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    //  配置消费者组(组名任意起名)

    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test666");



    // 创建消费者

    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

    List<String> topics = Collections.singletonList("testtopic");

    kafkaConsumer.subscribe(topics);

    // 拉取数据打印

    while (true){

        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(2));

        for (ConsumerRecord<String, String> msg : consumerRecords) {

            System.out.println("偏移量： "+ msg.offset()+"\t分区："+msg.partition()+"\t消息key: "+msg.key()+"\t消息内容: "+msg.value());

        }

    }

}
java
public static void main(String[] args) throws InterruptedException {

    Properties properties = new Properties();

    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.101.105:9092");

    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    //  配置消费者组(组名任意起名)

    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test666");



    // 创建消费者

    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

    List<String> topics = Collections.singletonList("testtopic");

    kafkaConsumer.subscribe(topics);

    // 拉取数据打印

    while (true){

        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(2));

        for (ConsumerRecord<String, String> msg : consumerRecords) {

            System.out.println("偏移量： "+ msg.offset()+"\t分区："+msg.partition()+"\t消息key: "+msg.key()+"\t消息内容: "+msg.value());

        }

    }

}
java
public static void main(String[] args) throws InterruptedException {

    Properties properties = new Properties();

    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.101.105:9092");

    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    //  配置消费者组(组名任意起名)

    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test666");



    // 创建消费者

    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

    List<String> topics = Collections.singletonList("testtopic");

    kafkaConsumer.subscribe(topics);

    // 拉取数据打印

    while (true){

        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(2));

        for (ConsumerRecord<String, String> msg : consumerRecords) {

            System.out.println("偏移量： "+ msg.offset()+"\t分区："+msg.partition()+"\t消息key: "+msg.key()+"\t消息内容: "+msg.value());

        }

    }

}
::: tip Kafka默认的分区分配策略就是`Range` + `CooperativeSticky`，所以不需要修改策略。 ::: 3. 启动CustomProducer生产者，发送 500 条消息，随机发送到不同的分区。 ```java public static void main(String[] args) throws InterruptedException { Properties properties = new Properties(); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.101.105:9092"); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); 
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
for (int i = 0; i < 500; i++) {
    int finalI = i;
    // 指定key值有助于打散消息到不同分区
    kafkaProducer.send(new ProducerRecord<>("testtopic",null, i+"", "我是spark"+i), new Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if(exception==null){
                System.out.println("当前"+ finalI +"消息发送完成, 所在分区："+metadata.partition());
            }else{
                exception.printStackTrace();
            }
        }
    });
    Thread.sleep(10);
}
kafkaProducer.close();

}

txt
运行结果： 
![Alt text](/bigdata/kafka/images/20240318113234.png)
![Alt text](/bigdata/kafka/images/20240318114613.png)
### 2.3 Range分区分配再平衡
1. 如图可以看到实际消费者分配分区情况如下：    
    - CustomGroupConsumer消费者：消费到 5、6 号分区数据。   
    - CustomGroup1Consumer消费者：消费到 0、1、2 号分区数据。  
    - CustomGroup2Consumer消费者：消费到 3、4 号分区数据。  
2. 停止掉CustomGroup1Consumer消费者，快速重新发送消息观看结果(45s 以内，越快越好):
CustomGroup1Consumer消费者的任务会整体被分配到CustomGroupConsumer消费者或者CustomGroup2Consumer消费者。  
![Alt text](/bigdata/kafka/images/20240318121601.png) 
:::tip
CustomGroup1Consumer消费者挂掉后，消费者组需要按照超时时间45s来判断它是否退出，所以需要等待，时间到了45s后，判断它真的退出就会把任务分配给其他 broker执行。
:::
3. 再次重新发送消息观看结果(45s 以后)   
- CustomGroup2Consumer消费者：消费到0、1、2、3号分区数据。  
- CustomGroupConsumer消费者：消费到4、5、6号分区数据。 
![Alt text](/bigdata/kafka/images/20240318121946.png)
:::tip
消费者CustomGroup1Consumer已经被踢出消费者组，所以重新按照Range方式分配。
:::
## 3. RoundRobin策略以及再平衡
### 3.1 RoundRobin分区策略原理
![Alt text](/bigdata/kafka/images/20240318105020.png)
### 3.2 RoundRobin分区代码实操
1. 依次在CustomGroupConsumer、CustomGroup1Consumer、CustomGroup2Consumer三个消费者代码中修改分区分配策略为RoundRobin:   
<div class="vp-code-group"><div class="tabs"><input type="radio" name="cg18" id="cg18-0" checked><label for="cg18-0">java [CustomGroupConsumer.java] {7}</label><input type="radio" name="cg18" id="cg18-1"><label for="cg18-1">java [CustomGroup1Consumer.java]{7}</label><input type="radio" name="cg18" id="cg18-2"><label for="cg18-2">java [CustomGroup2Consumer.java]{7}</label></div><div class="blocks"><div class="language-java [CustomGroupConsumer.java] {7} vp-adaptive-theme"><span class="lang">java [CustomGroupConsumer.java] {7}</span><pre class="shiki "><code><span class="line">public static void main(String[] args) {
</span>
<span class="line">    Properties properties = new Properties();
</span>
<span class="line">    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;192.168.101.105:9092&quot;);
</span>
<span class="line">    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
</span>
<span class="line">    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
</span>
<span class="line">    // 修改分区分配策略
</span>
<span class="line">    properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName());
</span>
<span class="line">    //  配置消费者组(组名任意起名)
</span>
<span class="line">    properties.put(ConsumerConfig.GROUP_ID_CONFIG, &quot;test666&quot;);
</span>
<span class="line">
</span>
<span class="line">    // 创建消费者
</span>
<span class="line">    KafkaConsumer&lt;String, String&gt; kafkaConsumer = new KafkaConsumer&lt;String, String&gt;(properties);
</span>
<span class="line">    List&lt;String&gt; topics = Collections.singletonList(&quot;testtopic&quot;);
</span>
<span class="line">    kafkaConsumer.subscribe(topics);
</span>
<span class="line">    // 拉取数据打印
</span>
<span class="line">    while (true){
</span>
<span class="line">        ConsumerRecords&lt;String, String&gt; consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(2));
</span>
<span class="line">        for (ConsumerRecord&lt;String, String&gt; msg : consumerRecords) {
</span>
<span class="line">            System.out.println(&quot;偏移量： &quot;+ msg.offset()+&quot;\t分区：&quot;+msg.partition()+&quot;\t消息key: &quot;+msg.key()+&quot;\t消息内容: &quot;+msg.value());
</span>
<span class="line">        }
</span>
<span class="line">    }
</span>
<span class="line">}
</span></code></pre></div><div class="language-java [CustomGroup1Consumer.java]{7} vp-adaptive-theme"><span class="lang">java [CustomGroup1Consumer.java]{7}</span><pre class="shiki "><code><span class="line">public static void main(String[] args) throws InterruptedException {
</span>
<span class="line">    Properties properties = new Properties();
</span>
<span class="line">    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;192.168.101.105:9092&quot;);
</span>
<span class="line">    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
</span>
<span class="line">    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
</span>
<span class="line">     // 修改分区分配策略
</span>
<span class="line">    properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName());
</span>
<span class="line">    //  配置消费者组(组名任意起名)
</span>
<span class="line">    properties.put(ConsumerConfig.GROUP_ID_CONFIG, &quot;test666&quot;);
</span>
<span class="line">
</span>
<span class="line">    // 创建消费者
</span>
<span class="line">    KafkaConsumer&lt;String, String&gt; kafkaConsumer = new KafkaConsumer&lt;String, String&gt;(properties);
</span>
<span class="line">    List&lt;String&gt; topics = Collections.singletonList(&quot;testtopic&quot;);
</span>
<span class="line">    kafkaConsumer.subscribe(topics);
</span>
<span class="line">    // 拉取数据打印
</span>
<span class="line">    while (true){
</span>
<span class="line">        ConsumerRecords&lt;String, String&gt; consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(2));
</span>
<span class="line">        for (ConsumerRecord&lt;String, String&gt; msg : consumerRecords) {
</span>
<span class="line">            System.out.println(&quot;偏移量： &quot;+ msg.offset()+&quot;\t分区：&quot;+msg.partition()+&quot;\t消息key: &quot;+msg.key()+&quot;\t消息内容: &quot;+msg.value());
</span>
<span class="line">        }
</span>
<span class="line">    }
</span>
<span class="line">}
</span></code></pre></div><div class="language-java [CustomGroup2Consumer.java]{7} vp-adaptive-theme"><span class="lang">java [CustomGroup2Consumer.java]{7}</span><pre class="shiki "><code><span class="line">public static void main(String[] args) throws InterruptedException {
</span>
<span class="line">    Properties properties = new Properties();
</span>
<span class="line">    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;192.168.101.105:9092&quot;);
</span>
<span class="line">    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
</span>
<span class="line">    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
</span>
<span class="line">     // 修改分区分配策略
</span>
<span class="line">    properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName());
</span>
<span class="line">    //  配置消费者组(组名任意起名)
</span>
<span class="line">    properties.put(ConsumerConfig.GROUP_ID_CONFIG, &quot;test666&quot;);
</span>
<span class="line">
</span>
<span class="line">    // 创建消费者
</span>
<span class="line">    KafkaConsumer&lt;String, String&gt; kafkaConsumer = new KafkaConsumer&lt;String, String&gt;(properties);
</span>
<span class="line">    List&lt;String&gt; topics = Collections.singletonList(&quot;testtopic&quot;);
</span>
<span class="line">    kafkaConsumer.subscribe(topics);
</span>
<span class="line">    // 拉取数据打印
</span>
<span class="line">    while (true){
</span>
<span class="line">        ConsumerRecords&lt;String, String&gt; consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(2));
</span>
<span class="line">        for (ConsumerRecord&lt;String, String&gt; msg : consumerRecords) {
</span>
<span class="line">            System.out.println(&quot;偏移量： &quot;+ msg.offset()+&quot;\t分区：&quot;+msg.partition()+&quot;\t消息key: &quot;+msg.key()+&quot;\t消息内容: &quot;+msg.value());
</span>
<span class="line">        }
</span>
<span class="line">    }
</span>
<span class="line">}
</span></code></pre></div></div></div>
3. 使用同样的生产者发送 500 条消息, 查看分区结果    
![分区结果图](/bigdata/kafka/images/20240318125013.png)
### 3.3 RoundRobin分区分配再平衡
1. 如图可以看到实际消费者分配分区情况如下：    
    - CustomGroupConsumer消费者：消费到0、3、6号分区数据。   
    - CustomGroup1Consumer消费者：消费到1、4号分区数据。  
    - CustomGroup2Consumer消费者：消费到2、5号分区数据。  
2. 停止掉CustomGroupConsumer消费者，快速重新发送消息观看结果(45s 以内，越快越好):    
消费者的任务会按照RoundRobin的方式，把数据轮询分成0、3和6号分区数据，分别由CustomGroup1Consumer消费者或者CustomGroup2Consumer消费者消费。  
![Alt text](/bigdata/kafka/images/20240318125708.png)
:::tip
CustomGroup1Consumer消费者挂掉后，消费者组需要按照超时时间45s来判断它是否退出，所以需要等待，时间到了45s后，判断它真的退出就会把任务分配给其他broker执行。  
:::
3. 再次重新发送消息观看结果(45s以后)  
![Alt text](/bigdata/kafka/images/20240318125950.png) 
:::tip
消费者CustomGroupConsumer已经被踢出消费者组，所以重新按照RoundRobin方式分配。
:::
## 4. Sticky以及再平衡
### 4.1 粘性分区定义
可以理解为分配的结果带有"粘性的"。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。   
粘性分区是Kafka从0.11.x版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。  
### 4.2 Sticky分区代码实操
1. 需求:   
设置主题为testtopic，7个分区；准备3个消费者，采用粘性分区策略，并进行消费，观察消费分配情况。然后再停止其中一个消费者，再次观察消费分配情况。
2. 修改CustomGroupConsumer、CustomGroup1Consumer、CustomGroup2Consumer配置分区分配策略为粘性
```java
// 修改分区分配策略
ArrayList<String> startegys = new ArrayList<>();
startegys.add("org.apache.kafka.clients.consumer.StickyAssignor");
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, tartegys);

信息

3个消费者如果启动出现报错，全部停止等会再重启，或者修改为全新的消费者组。

  1. 使用同样的生产者发送500条消息，查看分区结果
    Alt text 可以看到Kafka使用Sticky分区策略会尽量保持分区的个数近似划分分区。

4.3 Sticky分区分配再平衡

  1. 可以看到目前实际消费者分配分区情况如下：
    • CustomGroupConsumer消费者：消费到4、5号分区数据。
    • CustomGroup1Consumer消费者：消费到0、1、2号分区数据。
    • CustomGroup2Consumer消费者：消费到3、6号分区数据。
  2. 停止掉CustomGroupConsumer消费者，快速重新发送消息观察结果(45s以内，越快越好):
    Alt text
    CustomGroupConsumer消费者的任务会按照粘性规则，尽可能均衡的随机分成4和5号分区数据，分别由CustomGroup1Consumer消费者和CustomGroup2Consumer消费者消费。
  3. 再次重新发送消息观看结果(45s以后)
    • CustomGroup1Consumer消费者：消费到0、1、5号分区数据。
    • CustomGroup2Consumer消费者：消费到3、4、6号分区数据。
      Alt text 说明：消费者CustomGroup2Consumer已经被踢出消费者组，所以重新按照粘性方式分配。
最后更新于: 2025-06-04 07:53

评论