一、RoundRobin 分區(qū)分配策略原理

• RoundRobin 針對集群中所有Topic而言。
• RoundRobin 輪詢分區(qū)策略，是把所有的 partition 和所有的consumer 都列出來，然后按照 hashcode 進(jìn)行排序，最后通過輪詢算法來分配 partition 給到各個消費(fèi)者。
  

二、RoundRobin分區(qū)分配策略代碼案例

2.1、創(chuàng)建帶有7個分區(qū)的sixTopic主題

• 在 Kafka 集群控制臺，創(chuàng)建帶有7個分區(qū)的sixTopic主題

  bin/kafka-topics.sh --bootstrap-server 192.168.136.27:9092 --create --partitions 7 --replication-factor 1 --topic sixTopic
  

  

2.3、創(chuàng)建三個消費(fèi)者 組成 消費(fèi)者組

• 復(fù)制 CustomConsumer1類，創(chuàng)建 CustomConsumer2和CustomConsumer3。這樣可以由三個消費(fèi)者組成消費(fèi)者組，組名都為“test1”，設(shè)置分區(qū)分配策略為 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());
  
          // 配置消費(fèi)者組id
          properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test1");
          // 設(shè)置分區(qū)分配策略
          properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
  
          // 1 創(chuàng)建一個消費(fèi)者  "", "hello"
          KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
  
          // 2 訂閱主題 sixTopic
          ArrayList<String> topics = new ArrayList<>();
          topics.add("sixTopic");
          kafkaConsumer.subscribe(topics);
  
          // 3 消費(fèi)數(shù)據(jù)
          while (true){
  
              ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
  
              for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                  System.out.println(consumerRecord);
              }
          }
      }
  }
  

2.3、創(chuàng)建生產(chǎn)者

• 創(chuàng)建CustomProducer生產(chǎn)者。

  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、創(chuàng)建 kafka 生產(chǎn)者的配置對象
          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、指定對應(yīng)的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、創(chuàng)建 kafka 生產(chǎn)者對象
          KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
  
          //5、調(diào)用 send 方法,發(fā)送消息
          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() + " 分區(qū)： "+ metadata.partition());
                      }
                  }
              });
              Thread.sleep(2);
          }
  
          // 3 關(guān)閉資源
          kafkaProducer.close();
      }
  }
  
  

2.4、測試

• 首先，在 IDEA中分別啟動消費(fèi)者1、消費(fèi)者2和消費(fèi)者3代碼
  

• 然后，在 IDEA中分別啟動生產(chǎn)者代碼
  

• 在 IDEA 控制臺觀察消費(fèi)者1、消費(fèi)者2和消費(fèi)者3控制臺接收到的數(shù)據(jù)，如下圖所示：

  
  
  

2.5、RoundRobin分區(qū)分配策略代碼案例說明

• 由上述測試輸出結(jié)果截圖可知： 消費(fèi)者1消費(fèi)1、4分區(qū)的數(shù)據(jù)；消費(fèi)者2消費(fèi)2和5分區(qū)的數(shù)據(jù)；消費(fèi)者3消費(fèi)0、3、6分區(qū)的數(shù)據(jù)。
• 說明：Kafka 采用修改后的RoundRobin分區(qū)分配策略。

三、RoundRobin 分區(qū)分配再平衡案例

3.1、停止某一個消費(fèi)者后，（45s 以內(nèi)）重新發(fā)送消息示例

• 由下圖控制臺輸出可知：2號消費(fèi)者 消費(fèi)到 2、5號分區(qū)數(shù)據(jù)。
  
• 由下圖控制臺輸出可知：3號消費(fèi)者 消費(fèi)到 0、3、6號分區(qū)數(shù)據(jù)。
  

3.2、停止某一個消費(fèi)者后，（45s 以后）重新發(fā)送消息示例

• 由下圖控制臺輸出可知：2號消費(fèi)者 消費(fèi)到 1、3、5號分區(qū)數(shù)據(jù)。
  

• 由下圖控制臺輸出可知：3號消費(fèi)者 消費(fèi)到 0、2、4、6號分區(qū)數(shù)據(jù)。
  文章來源地址http://www.zghlxwxcb.cn/news/detail-733573.html

3.3、RoundRobin 分區(qū)分配再平衡案例說明

• 1號消費(fèi)者掛掉后，消費(fèi)者組需要按照超時時間 45s 來判斷它是否退出，所以需要等待，時間到了 45s 后，判斷它真的退出就會把任務(wù)分配給其他 broker 執(zhí)行。
• 消費(fèi)者 1 已經(jīng)被踢出消費(fèi)者組，所以重新按照 RoundRobin 方式分配。

到了這里，關(guān)于Kafka3.0.0版本——消費(fèi)者（RoundRobin分區(qū)分配策略以及再平衡）的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！