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

心已赠人 2024-03-04 01:51 125阅读 0赞

#### 目录 ####

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

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

*  **RoundRobin 针对集群中所有Topic而言。**
 *  **RoundRobin 轮询分区策略，是把所有的 partition 和所有的consumer 都列出来，然后按照 hashcode 进行排序，最后通过轮询算法来分配 partition 给到各个消费者。**  
    ![在这里插入图片描述][12a5d358175c43aca28890d904a1ae60.png]

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

#### 2.1、创建带有7个分区的sixTopic主题 ####

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

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

*  **复制 CustomConsumer1类，创建 CustomConsumer2和CustomConsumer3。这样可以由三个消费者组成消费者组，组名都为“test1”，设置分区分配策略为 RoundRobin。**
    
        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,"test1");
                // 设置分区分配策略
                properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
        
                // 1 创建一个消费者  "", "hello"
                KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        
                // 2 订阅主题 sixTopic
                ArrayList<String> topics = new ArrayList<>();
                topics.add("sixTopic");
                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<>("sixTopic", "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代码**  
    ![在这里插入图片描述][dc4cbbd65e2e49558703ee477a92f6c1.png]
 *  **然后，在 IDEA中分别启动生产者代码**  
    ![在这里插入图片描述][152dd75b4e2145b7865f94877045f2ea.png]
 *  **在 IDEA 控制台观察消费者1、消费者2和消费者3控制台接收到的数据，如下图所示：**
    
    ![在这里插入图片描述][4b6094ae198f4cc1a59a566cd39b420f.png]  
    ![在这里插入图片描述][7bb8fd7fbb0244c0b0fe55fae6de022b.png]  
    ![在这里插入图片描述][f708027ab7a14376941acd2743f74442.png]

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

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

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

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

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

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

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

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

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

[12a5d358175c43aca28890d904a1ae60.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/a05771ebcaa64d91b9de0690dd518c40.png
[064c6429f43348f4a162a8d405e85812.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/2376a44a1ce04e13b7af474510ddda10.png
[dc4cbbd65e2e49558703ee477a92f6c1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/70751e526c5e438a94829db2c7b5f1bb.png
[152dd75b4e2145b7865f94877045f2ea.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/2798646d96ac4725b55c500cc12db5ac.png
[4b6094ae198f4cc1a59a566cd39b420f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/e3349202c47440d1acb3b3f0ec1abe32.png
[7bb8fd7fbb0244c0b0fe55fae6de022b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/e42c08f38dd74bc9a394efc8ea81f7de.png
[f708027ab7a14376941acd2743f74442.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/13d11587dfae4f35b8c6b75f4656a8d5.png
[24491d612d1741c98875d3aeb3e39936.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/c9ce9ee463a64e4fb18c7194a969cb5c.png
[16fce78de5a24f2d845d31c9e1a6fc45.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/27c60a7d12124a20acc47196d1a86ddf.png
[a30924ac50284b38876266dfd2ee6763.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/eea0c25c07d843fe80f7fb991e99ed20.png
[7690c247a99a476e9361ba48ee654293.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/6569630567fa41a0b354bb92dedadf02.png