Kafka3.0.0版本——消费者（Range分区分配策略以及再平衡）

本是古典何须时尚 2024-03-04 03:17 58阅读 0赞

#### 目录 ####

*   *  一、Range分区分配策略原理
     *   *  1.1、Range分区分配策略原理的示例一
         *  1.2、Range分区分配策略原理的示例二
         *  1.3、Range分区分配策略原理的示例注意事项
     *  二、Range 分区分配策略代码案例
     *   *  2.1、创建带有4个分区的fiveTopic主题
         *  2.2、创建三个消费者 组成 消费者组
         *  2.3、创建生产者
         *  2.4、测试
         *  2.5、Range 分区分配策略代码案例说明
     *  三、Range 分区分配再平衡案例
     *   *  3.1、停止某一个消费者后，（45s 以内）重新发送消息示例
         *  3.2、停止某一个消费者后，（45s 以后）重新发送消息示例
         *  3.3、Range 分区分配再平衡案例说明

### 一、Range分区分配策略原理 ###

*  **Range 是对每个 topic 而言的。首先对同一个 topic 里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。**

#### 1.1、Range分区分配策略原理的示例一 ####

> 假如现在有 4 个分区，3 个消费者，排序后的分区将会是0,1,2,3；消费者排序完之后将会是C1,C2,C3。

*  **通过 partitions数/consumer数 来决定每个消费者应该消费几个分区。 如果除不尽，那么前面几个消费者将会多消费 1 个分区。**
 *  **例如：4/3 = 1 余 1 ，除不尽，那么消费者C1便会多消费1个分区。**  
    ![在这里插入图片描述][b59a8c3979ee43a9a88f1f905b53cae9.png]

#### 1.2、Range分区分配策略原理的示例二 ####

> 假如现在有 5 个分区，3 个消费者，排序后的分区将会是0,1,2,3,4；消费者排序完之后将会是C1,C2,C3。

*  **通过 partitions数/consumer数 来决定每个消费者应该消费几个分区。 如果除不尽，那么前面几个消费者将会多消费 1 个分区。**
 *  **例如：5/3 = 1 余 2 ，除不尽，那么消费者么C1和C2分别多消费一个分区。**  
    ![在这里插入图片描述][49f76f2cef3f437f913eae924468f74b.png]

#### 1.3、Range分区分配策略原理的示例注意事项 ####

*  **如果只是针对 1 个 topic 而言，C1消费者多消费1个分区影响不是很大。但是如果有N多个topic，那么针对每个 topic，消费者 C1都将多消费 1 个分区，topic越多，C1消费的分区会比其他消费者明显多消费 N 个分区。 容易产生数据倾斜！**

### 二、Range 分区分配策略代码案例 ###

#### 2.1、创建带有4个分区的fiveTopic主题 ####

*  **在 Kafka 集群控制台，创建带有4个分区的fiveTopic主题**
    
        bin/kafka-topics.sh --bootstrap-server 192.168.136.27:9092 --create --partitions 4 --replication-factor 1 --topic fiveTopic
    
    ![在这里插入图片描述][1bc81cd07f154cf8a97f19d564a5c8ee.png]

#### 2.2、创建三个消费者 组成 消费者组 ####

*  **复制 CustomConsumer1类，创建 CustomConsumer2和CustomConsumer3。这样可以由三个消费者组成消费者组，组名都为“test”。**
    
        package com.xz.kafka.consumer;
        
        import org.apache.kafka.clients.consumer.ConsumerConfig;
        import org.apache.kafka.clients.consumer.ConsumerRecord;
        import org.apache.kafka.clients.consumer.ConsumerRecords;
        import org.apache.kafka.clients.consumer.KafkaConsumer;
        import org.apache.kafka.common.serialization.StringDeserializer;
        
        import java.time.Duration;
        import java.util.ArrayList;
        import java.util.Properties;
        
        public class CustomConsumer1 {
              
        
            public static void main(String[] args) {
              
        
                // 0 配置
                Properties properties = new Properties();
        
                // 连接 bootstrap.servers
                properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092");
        
                // 反序列化
                properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
                properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        
                // 配置消费者组id
                properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        
                // 1 创建一个消费者  "", "hello"
                KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        
                // 2 订阅主题 first
                ArrayList<String> topics = new ArrayList<>();
                topics.add("fiveTopic");
                kafkaConsumer.subscribe(topics);
        
                // 3 消费数据
                while (true){
              
        
                    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
        
                    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
              
                        System.out.println(consumerRecord);
                    }
                }
            }
        }

#### 2.3、创建生产者 ####

*  **创建CustomProducer生产者。**
    
        package com.xz.kafka.producer;
        
        import org.apache.kafka.clients.producer.*;
        import org.apache.kafka.common.serialization.StringSerializer;
        import java.util.Properties;
        
        public class CustomProducerCallback {
              
        
            public static void main(String[] args) throws InterruptedException {
              
        
                //1、创建 kafka 生产者的配置对象
                Properties properties = new Properties();
        
                //2、给 kafka 配置对象添加配置信息：bootstrap.servers
                properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092");
        
                //3、指定对应的key和value的序列化类型 key.serializer value.serializer
                properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
                properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        
                //4、创建 kafka 生产者对象
                KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        
                //5、调用 send 方法,发送消息
                for (int i = 0; i < 200; i++) {
              
                    kafkaProducer.send(new ProducerRecord<>("fiveTopic", "hello kafka" + i), new Callback() {
              
                        @Override
                        public void onCompletion(RecordMetadata metadata, Exception exception) {
              
                            if (exception == null){
              
                                System.out.println("主题： "+metadata.topic() + " 分区： "+ metadata.partition());
                            }
                        }
                    });
                    Thread.sleep(2);
                }
        
                // 3 关闭资源
                kafkaProducer.close();
            }
        }

#### 2.4、测试 ####

*  **首先，在 IDEA中分别启动消费者1、消费者2和消费者3代码**  
    ![在这里插入图片描述][ef14d10b33b24d9d837091c2fb5b2f28.png]
 *  **然后，在 IDEA中分别启动生产者代码**  
    ![在这里插入图片描述][8a8e5e2ba4aa4334bb89788032bb1f39.png]
 *  **在 IDEA 控制台观察消费者1、消费者2和消费者3控制台接收到的数据，如下图所示：**
    
    ![在这里插入图片描述][85a6e8ae7b4146f59d42c0a29530bc55.png_pic_center]  
    ![在这里插入图片描述][95b5ddb96c5149a79cd8636a60cddc22.png_pic_center]  
    ![在这里插入图片描述][4cfe0beaf785421ab13e65d3e1a2fa84.png_pic_center]

#### 2.5、Range 分区分配策略代码案例说明 ####

*  **由上述测试输出结果截图可知： 消费者1消费2分区的数据；消费者2消费0和3分区的数据；消费者3消费2分区的数据。**
 *  **说明：Kafka 默认的分区分配策略就是 Range + CooperativeSticky，所以不需要修改策略。**

### 三、Range 分区分配再平衡案例 ###

#### 3.1、停止某一个消费者后，（45s 以内）重新发送消息示例 ####

*  **由下图控制台输出可知：2号消费者 消费到 0、3号分区数据。**![在这里插入图片描述][4e611f480323481aada5c07cbfbdcb9b.png_pic_center]
 *  **由下图控制台输出可知：3号消费者 消费到 1号分区数据。**  
    ![在这里插入图片描述][11be84af69104718a949d11f2b598aea.png_pic_center]

#### 3.2、停止某一个消费者后，（45s 以后）重新发送消息示例 ####

*  **由下图控制台输出可知：2号消费者 消费到 0、3号分区数据。**  
    ![在这里插入图片描述][5c8a7c2d1e5e4e85a20b3f9bc5f7e7c4.png]
 *  **由下图控制台输出可知：3号消费者 消费到 1、2号分区数据。**  
    ![在这里插入图片描述][7d4643caa5044949bc791bf65cb9144f.png]

#### 3.3、Range 分区分配再平衡案例说明 ####

*  **1号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。**
 *  **消费者1 已经被踢出消费者组，所以重新按照 range 方式分配。**

[b59a8c3979ee43a9a88f1f905b53cae9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/4f57b80169bc48178515db8bd9fde9c1.png
[49f76f2cef3f437f913eae924468f74b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/e4c6935ce4ea44f19edd7096fb4a4404.png
[1bc81cd07f154cf8a97f19d564a5c8ee.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/b9ac8395971e4cb58cd0ee36ff55abb5.png
[ef14d10b33b24d9d837091c2fb5b2f28.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/1677ccaef44b49098cbe3cb59c34e47c.png
[8a8e5e2ba4aa4334bb89788032bb1f39.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/33c50c76482945ec91a969125e76dde4.png
[85a6e8ae7b4146f59d42c0a29530bc55.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/79076d4ec18348508340edb53806b707.png
[95b5ddb96c5149a79cd8636a60cddc22.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/f4a62f1a0b5342d9833d400f0b549b77.png
[4cfe0beaf785421ab13e65d3e1a2fa84.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/b641436a892d457fa888a880a8ab04f5.png
[4e611f480323481aada5c07cbfbdcb9b.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/bd0bb9aa18c64bcca44602b27bf4964e.png
[11be84af69104718a949d11f2b598aea.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/6213b27b37a74226afb7352d7da62899.png
[5c8a7c2d1e5e4e85a20b3f9bc5f7e7c4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/0f637d4cf484430e82da85a4ce5ef471.png
[7d4643caa5044949bc791bf65cb9144f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/3faf33fdc4c34e7d814454b8972478cb.png