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

快来打我* 2024-03-04 01:27 117阅读 0赞

#### 目录 ####

*   *  一、Sticky分区分配策略原理
     *  二、Sticky分区分配策略 示例需求
     *  三、Sticky分区分配策略代码案例
     *   *  3.1、创建带有7个分区的sevenTopic主题
         *  3.2、创建三个消费者 组成 消费者组
         *  3.3、创建生产者
         *  3.4、测试
         *  3.5、Sticky分区分配策略代码案例说明
     *  四、Sticky分区分配再平衡案例
     *   *  4.1、停止某一个消费者后，（45s 以内）重新发送消息示例
         *  4.2、停止某一个消费者后，（45s 以后）重新发送消息示例
         *  4.3、Sticky分区分配再平衡案例说明

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

*  **粘性分区定义：可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销。**
 *  **粘性分区是 Kafka 从 0.11.x 版本开始引入这种分配策略，首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。**

### 二、Sticky分区分配策略 示例需求 ###

*  **设置主题为 sevenTopic，7个分区；准备 3 个消费者，采用粘性分区策略，并进行消费，观察  
    消费分配情况。然后再停止其中一个消费者，再次观察消费分配情况。**

### 三、Sticky分区分配策略代码案例 ###

#### 3.1、创建带有7个分区的sevenTopic主题 ####

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

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

*  **复制 CustomConsumer1类，创建 CustomConsumer2和CustomConsumer3。这样可以由三个消费者组成消费者组，组名都为“test2”，设置分区分配策略为 Sticky。**
    
        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,"test2");
                // 设置分区分配策略
                properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");
        
                // 1 创建一个消费者  "", "hello"
                KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        
                // 2 订阅主题 sevenTopic
                ArrayList<String> topics = new ArrayList<>();
                topics.add("sevenTopic");
                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);
                    }
                }
            }
        }

#### 3.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<>("sevenTopic", "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();
            }
        }

#### 3.4、测试 ####

*  **首先，在 IDEA中分别启动消费者1、消费者2和消费者3代码**  
    ![在这里插入图片描述][dc4cbbd65e2e49558703ee477a92f6c1.png]
 *  **然后，在 IDEA中分别启动生产者代码**  
    ![在这里插入图片描述][c32c5a22a65a417eba81b93e2848a854.png]
 *  **在 IDEA 控制台观察消费者1、消费者2和消费者3控制台接收到的数据，如下图所示：**  
    ![在这里插入图片描述][9dddf0c534de478d865e0de638433dae.png]  
    ![在这里插入图片描述][6f89d7cddf374e15b304680e4371696b.png]  
    ![在这里插入图片描述][063940765e1b49feb03bc98f4bef8cdf.png]

#### 3.5、Sticky分区分配策略代码案例说明 ####

*  **由上述测试输出结果截图可知： 消费者1消费0、2、4分区的数据；消费者2消费1、3分区的数据；消费者3消费5、6分区的数据。**
 *  **说明：Kafka 采用修改后的Sticky分区分配策略。**

### 四、Sticky分区分配再平衡案例 ###

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

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

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

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

#### 4.3、Sticky分区分配再平衡案例说明 ####

*  **1 号消费者的任务会按照粘性规则，尽可能均衡的随机分成 0 和 1 号分区数据，分别由 2号消费者或者 3号消费者消费。**
 *  **1号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。**
 *  **消费者 1 已经被踢出消费者组，所以重新按照粘性方式分配。**

[9657eee6a05a4673a5bfd507b7e0970d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/8c8301fac8504ea1b9de9e4e6e2f239f.png
[dc4cbbd65e2e49558703ee477a92f6c1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/457289e2c2dc48a6902528fe8f22edfc.png
[c32c5a22a65a417eba81b93e2848a854.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/8adca98fd640423d8c9e04c9637f373d.png
[9dddf0c534de478d865e0de638433dae.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/9b092f2e36ff4cacbcbd0015d451205c.png
[6f89d7cddf374e15b304680e4371696b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/c58fcf8a156c4bf791b485f0a7527c07.png
[063940765e1b49feb03bc98f4bef8cdf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/c1bba91db6d64c29842cad460941e11e.png
[6af117612deb44318e437b881aa15c22.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/cdbdba8066bc44a895c948f7d24934db.png
[3b076b12128b4b3f86366fd680c33d48.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/04f2df0c8341483bac0d5d4ef64203c0.png
[78eaaebf31a94e16ad9e49f434aaefe9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/1e2867f7927b4cf79333bf4db3bd4069.png
[94e00109a41e4e369cd2ac748762f5e9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/fa9cb7c399d74cacbb00145558e13025.png