kafka rebalance 機制與Consumer多種消費模式案例應用實戰-kafka 商業環境實戰

本套系列博客從真實商業環境抽取案例進行總結和分享，並給出Spark商業應用實戰指導，請持續關注本套博客。版權聲明：本套Spark商業應用實戰歸做者（秦凱新）全部，禁止轉載，歡迎學習。

• kafka 商業環境實戰-kafka生產環境規劃
• kafka 商業環境實戰-kafka生產者和消費者吞吐量測試
• kafka 商業環境實戰-kafka生產者Producer參數設置及參數調優建議
• kafka 商業環境實戰-kafka集羣管理重要操做指令運維兵書
• kafka 商業環境實戰-kafka集羣Broker端參數設置及調優準則建議
• kafka 商業環境實戰-kafka之Producer同步與異步消息發送及事務冪等性案例應用實戰
• kafka 商業環境實戰-kafka Poll輪詢機制與消費者組的重平衡分區策略剖析
• kafka 商業環境實戰-kafka rebalance 機制與Consumer多種消費模式案例應用實戰
• [kafka 商業環境實戰-kafka副本與ISR同步機制原理深刻剖析]
• [kafka 商業環境實戰-kafka精確一次語義EOS的原理深刻剖析]
• [kafka 商業環境實戰-kafka消息的冪等性與事務支持機制深刻剖析]
• [kafka 商業環境實戰-kafka集羣Controller競選與責任設計思路架構詳解]
• [kafka 商業環境實戰-kafka集羣消息格式之V1版本到V2版本的平滑過渡詳解]
• [kafka 商業環境實戰-kafka集羣水印與leader epoch對數據一致性保障的深刻研究]
• [kafka 商業環境實戰-kafka集羣日誌文件系統設計與留存機制及Compact深刻研究]
• [kafka 商業環境實戰-kafka集羣Consumer group狀態機及Coordinaor管理機制深刻剖析]
• [kafka 商業環境實戰-kafka調優過程在吞吐量，持久性，低延時，可用性等指標的折中選擇研究]

1 rebalance 什麼時候觸發？到底幹嗎？流程如何？

1.1 reblance 什麼時候觸發

• 組訂閱發生變動，好比基於正則表達式訂閱，當匹配到新的topic建立時，組的訂閱就會發生變動。
• 組的topic分區數發生變動，經過命令行腳本增長了訂閱topic的分區數。
• 組成員發生變動：新加入組以及離開組。

1.2 reblance 到底幹嗎

一句話：多個Consumer訂閱了一個Topic時，根據分區策略進行消費者訂閱分區的重分配

1.3 Coordinator 到底在那個Broker

找到Coordinator的算法 與 找到_consumer_offsets目標分區的算法是一致的。

• 第一步：肯定目標分區：Math.abs(groupId.hashCode)%50，假設是12。
• 第二步：找到_consumer_offsets分區爲10的Leader副本所在的Broker，那麼該broker即爲Group Coordinator。

1.4 reblance 流程如何

reblance 流程流程總體以下圖所示，值得強調的幾點以下：

• Coordinator的角色由Broker端擔任。
• Group Leader 的角色主要有Consumer擔任。
• 加入組請求（JoinGroup）=>做用在於選擇Group Leader。
• 同步組請求（SyncGroup）=>做用在於肯定分區分配方案給Coordinator，把方案響應給全部Consumer。

  

1.5 reblance 機制的好處

• 分區分配權利下放給客戶端consumer，所以系統不用重啓，既能夠實現分區策略的變動。
• 用戶能夠自行實現機架感知分配方案。

1.6 reblance generation 過濾無用請求

• kafka引入 reblance generation ，就是爲了防止Consumer group的無效Offset提交。若由於某些緣由，consumer延遲提交了Offset，而該consumer被踢出了消費組，那麼該Consumer再次提交位移時，攜帶的就是舊的generation了。

2 reblance 監聽器應用級別實戰

• reblance 監聽器解決用戶把位移提交到外部存儲的狀況，在監聽器中實現位移保存和位移的重定向。

• onPartitionsRevoked : rebalance開啓新一輪的重平衡前會調用，通常用於手動提交位移，及審計功能

• onPartitionsAssigned ：rebalance在重平衡結束後會調用，通常用於消費邏輯處理

  Properties props = new Properties();
     props.put("bootstrap.servers", "localhost:9092");
     props.put("group.id", "test");
     props.put("enable.auto.commit", "false");
     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
     
     統計rebalance總時長
     final AtomicLong totalRebalanceTimeMs =new AtomicLong(0L)
     
     統計rebalance開始時刻
     final AtomicLong rebalanceStart =new AtomicLong(0L)
     
     
     1 重平衡監聽
     consumer.subscribe(Arrays.asList("test-topic"), new ConsumerRebalanceListener(){
         
     @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            for(TopicPartition tp : partitions){
            
                1 保存到外部存儲
                saveToExternalStore(consumer.position(tp)) 
                
                2 手動提交位移
                //consumer.commitSync(toCommit);
            }
            
            rebalanceStart.set(System.currentTimeMillis())
        }
    }
  
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
  
        totalRebalanceTimeMs.addAndGet(System.currentTimeMillis()-rebalanceStart.get())
        
        for (TopicPartition tp : partitions) {
               
               consumer.seek(tp，readFromExternalStore(tp))
        }
    }
     });
  
    2 消息處理
    while (true) {
     ConsumerRecords<String, String> records = consumer.poll(100);
     for (ConsumerRecord<String, String> record : records) {
         buffer.add(record);
     }
     if (buffer.size() >= minBatchSize) {
         insertIntoDb(buffer);
         consumer.commitSync();
         buffer.clear();
     }
    }
  複製代碼

3 Consumer組內消息均衡實戰

3.1 Consumer 單線程封裝，實現多個消費者來消費（浪費資源）

實例主題：

• ConsumerGroup 實現組封裝
• ConsumerRunnable 每一個線程維護私有的KafkaConsumer實例

---

public class  Main {

    public static void main(String[] args) {
        String brokerList = "localhost:9092";
        String groupId = "testGroup1";
        String topic = "test-topic";
        int consumerNum = 3;

        核心對外封裝
        ConsumerGroup consumerGroup = new ConsumerGroup(consumerNum, groupId, topic, brokerList);
        consumerGroup.execute();
    }
}
複製代碼

---

import java.util.ArrayList;
import java.util.List;
public class ConsumerGroup {    

    private List<ConsumerRunnable> consumers;
    
    public ConsumerGroup(int consumerNum, String groupId, String topic, String brokerList) {
        consumers = new ArrayList<>(consumerNum);
        for (int i = 0; i < consumerNum; ++i) {
            ConsumerRunnable consumerThread = new ConsumerRunnable(brokerList, groupId, topic);
            consumers.add(consumerThread);
        }
    }
    public void execute() {
        for (ConsumerRunnable task : consumers) {
            new Thread(task).start();
        }
    }
}
複製代碼

---

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;

public class ConsumerRunnable implements Runnable {

    private final KafkaConsumer<String, String> consumer;

    public ConsumerRunnable(String brokerList, String groupId, String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");        //本例使用自動提交位移
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        this.consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topic));   // 本例使用分區副本自動分配策略
    }

    @Override
    public void run() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(200);   
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(Thread.currentThread().getName() + " consumed " + record.partition() +
                        "th message with offset: " + record.offset());
            }
        }
    }
}
複製代碼

---

3.2 一個Consumer，內部實現多線程消費（consumer壓力過大）

實例主題：

• ConsumerHandler 單一的Consumer實例，poll后里面會跑一個線程池，執行多個Processor線程來處理
• Processor 業務邏輯處理方法

進一步優化建議；

• ConsumerHandler 設置手動提交位移，負責最終位移提交consumer.commitSync();。
• ConsumerHandler設置一個全局的Map<TopicPartion,OffsetAndMetadata> offsets，來管理Processor消費的位移。
• Processor 負責批處理完消息後，獲得消息的最大位移，並更新offsets數組
• ConsumerHandler 根據 offsets，位移提交後會清空offsets集合。
• ConsumerHandler設置重平衡監聽

---

public class Main {
    public static void main(String[] args) {
    
        String brokerList = "localhost:9092,localhost:9093,localhost:9094";
        String groupId = "group2";
        String topic = "test-topic";
        int workerNum = 5;

        ConsumerHandler consumers = new ConsumerHandler(brokerList, groupId, topic);
        consumers.execute(workerNum);
        try {
            Thread.sleep(1000000);
        } catch (InterruptedException ignored) {}
        consumers.shutdown();
    }
}
複製代碼

---

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ConsumerHandler {

    private final KafkaConsumer<String, String> consumer;
    private ExecutorService executors;

    public ConsumerHandler(String brokerList, String groupId, String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topic));
    }
    public void execute(int workerNum) {
    
        executors = new ThreadPoolExecutor(workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
                
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(200);
            
            for (final ConsumerRecord record : records) {
                executors.submit(new Processor(record));
            }
            
            
        }
    }
    public void shutdown() {
        if (consumer != null) {
            consumer.close();
        }
        if (executors != null) {
            executors.shutdown();
        }
        try {
            if (!executors.awaitTermination(10, TimeUnit.SECONDS)) {
                System.out.println("Timeout.... Ignore for this case");
            }
        } catch (InterruptedException ignored) {
            System.out.println("Other thread interrupted this shutdown, ignore for this case.");
            Thread.currentThread().interrupt();
        }
    }

}
複製代碼

---

import org.apache.kafka.clients.consumer.ConsumerRecord;
public class Processor implements Runnable {

    private ConsumerRecord<String, String> consumerRecord;

    public Processor(ConsumerRecord record) {
        this.consumerRecord = record;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " consumed " + consumerRecord.partition()
            + "th message with offset: " + consumerRecord.offset());
    }
}   
複製代碼

3.3 方案對比

• 第一種方案：建議採用 Consumer 單線程封裝，實現多個消費者來消費（浪費資源），這樣能很好地保證分區內消費的順序，同時也沒有線程切換的開銷。
• 第二種方案：實現複雜，問題在於可能沒法維護分區內的消息順序，注意消息處理和消息接收解耦了。

4 Consumer指定分區消費案例實戰（Standalone Consumer）

• Standalone Consumer assign 用於接收指定分區列表的消息和Subscribe是矛盾的。只能二選一。

• 多個 Consumer 實例消費一個 Topic 藉助於 group reblance可謂是天做之合。

• 若要精準控制，assign逃不了。

  poperties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "false");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<>(props);
        
        List<TopicPartion> partitions = new ArrayList<>();
        
        List<PartitionInfo>  allPartitions = consumer.partitionsFor("kaiXinTopic")
        
        if(allPartitions != null && !allPartitions.isEmpty){
            for(PartitionInfo partitionInfo : allPartitions){
            
                partitions.add(new TopicPartition(partitionInfo.topic(),partitionInfo.partition()))
            }
            
            consumer.assign(partitions)
        }
        
        while (true) {
           ConsumerRecords<String, String> records = consumer.poll(100);
           for (ConsumerRecord<String, String> record : records) {
               buffer.add(record);
           }
           if (buffer.size() >= minBatchSize) {
               insertIntoDb(buffer);
               
               consumer.commitSync();
               
               buffer.clear();
        }
  複製代碼

結語

本文綜合了多本Kafka實戰書籍和博客，爲了寫好本文，參考了大量資料，進行了語言的重組，辛苦成文，各自珍惜！

秦凱新 2181119 2123