Kafka消費者-從Kafka讀取數據

（1）Customer和Customer Group

（1）兩種常用的消息模型

隊列模型（queuing）和發布-訂閱模型（publish-subscribe）。

隊列的處理方式是一組消費者從服務器讀取消息，一條消息只由其中的一個消費者來處理。

發布-訂閱模型中，消息被廣播給所有的消費者，接收到消息的消費者都可以處理此消息。

（2）Kafka的消費者和消費者組

Kafka為這兩種模型提供了單一的消費者抽象模型： 消費者組 （consumer group）。 消費者用一個消費者組名標記自己。 一個發布在Topic上消息被分發給此消費者組中的一個消費者。 假如所有的消費者都在一個組中，那么這就變成了隊列模型。 假如所有的消費者都在不同的組中，那么就完全變成了發布-訂閱模型。 一個消費者組中消費者訂閱同一個Topic，每個消費者接受Topic的一部分分區的消息，從而實現對消費者的橫向擴展，對消息進行分流。

注意：當單個消費者無法跟上數據生成的速度，就可以增加更多的消費者分擔負載，每個消費者只處理部分partition的消息，從而實現單個應用程序的橫向伸縮。但是不要讓消費者的數量多於partition的數量，此時多余的消費者會空閑。此外，Kafka還允許多個應用程序從同一個Topic讀取所有的消息，此時只要保證每個應用程序有自己的消費者組即可。

消費者組的概念就是：當有多個應用程序都需要從Kafka獲取消息時，讓每個app對應一個消費者組，從而使每個應用程序都能獲取一個或多個Topic的全部消息；在每個消費者組中，往消費者組中添加消費者來伸縮讀取能力和處理能力，消費者組中的每個消費者只處理每個Topic的一部分的消息，每個消費者對應一個線程。

 

（3）線程安全

在同一個群組中，無法讓一個線程運行多個消費者，也無法讓多線線程安全地共享一個消費者。按照規則，一個消費者使用一個線程，如果要在同一個消費者組中運行多個消費者，需要讓每個消費者運行在自己的線程中。最好把消費者的邏輯封裝在自己的對象中，然后使用java的ExecutorService啟動多個線程，使每個消費者運行在自己的線程上,可參考https://www.confluent.io/blog

 （2）Partition Rebalance分區再均衡

（1）消費者組中新添加消費者讀取到原本是其他消費者讀取的消息

（2）消費者關閉或崩潰之后離開群組，原本由他讀取的partition將由群組里其他消費者讀取

（3）當向一個Topic添加新的partition，會發生partition在消費者中的重新分配

以上三種現象會使partition的所有權在消費者之間轉移，這樣的行為叫作再均衡。

再均衡的優點：

給消費者組帶來了高可用性和伸縮性

再均衡的缺點：

（1）再均衡期間消費者無法讀取消息，整個群組有一小段時間不可用

（2）partition被重新分配給一個消費者時，消費者當前的讀取狀態會丟失，有可能還需要去刷新緩存，在它重新恢復狀態之前會拖慢應用程序。

因此需要進行安全的再均衡和避免不必要的再均衡。

（3）創建Kafka消費者、訂閱主題、輪詢

Properties props = new Properties();  
      props.put("bootstrap", "broker1:9092,broker2:9092");
      props.put("group.id", "CountryCounter");
      props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
      props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
      //1.創建消費者
      KafkaConsuner<String, String> consumer = new KafkaConsumer<String, String>(props);
      
      //2.訂閱Topic
      //創建一個只包含單個元素的列表，Topic的名字叫作customerCountries
      consumer.subscribe(Collections.singletonList("customerCountries"));
      //支持正則表達式，訂閱所有與test相關的Topic
      //consumer.subscribe("test.*");
      
      //3.輪詢
      //消息輪詢是消費者的核心API，通過一個簡單的輪詢向服務器請求數據，一旦消費者訂閱了Topic，輪詢就會處理所欲的細節，包括群組協調、partition再均衡、發送心跳
      //以及獲取數據，開發者只要處理從partition返回的數據即可。
      try {
          while (true) {//消費者是一個長期運行的程序，通過持續輪詢向Kafka請求數據。在其他線程中調用consumer.wakeup()可以退出循環
              //在100ms內等待Kafka的broker返回數據.超市參數指定poll在多久之后可以返回，不管有沒有可用的數據都要返回
              ConsumerRecord<String, String> records = consumer.poll(100);
              for (ConsumerRecord<String, String> record : records) {
                  log.debug(record.topic() + record.partition() + record.offset() + record.key() + record.value());
                  //統計各個地區的客戶數量，即模擬對消息的處理
                  int updatedCount = 1;
                  updatedCount += custCountryMap.getOrDefault(record.value(), 0) + 1;
                  custCountryMap.put(record.value(), updatedCount);
                  
                  //真實場景中，結果一般會被保存到數據存儲系統中
                  JSONObject json = new JSONObject(custCountryMap);
                  System.out.println(json.toString(4));
              }
          }
      } finally {
        //退出應用程序前使用close方法關閉消費者，網絡連接和socket也會隨之關閉，並立即觸發一次再均衡
          consumer.close();
      }

（4）消費者的配置

1：fetch.min.bytes，指定消費者從broker獲取消息的最小字節數，即等到有足夠的數據時才把它返回給消費者

2：fetch.max.wait.ms，等待broker返回數據的最大時間，默認是500ms。fetch.min.bytes和fetch.max.wait.ms哪個條件先得到滿足，就按照哪種方式返回數據

3：max.partition.fetch.bytes，指定broker從每個partition中返回給消費者的最大字節數，默認1MB

4：session.timeout.ms，指定消費者被認定死亡之前可以與服務器斷開連接的時間，默認是3s

5：auto.offset.reset，消費者在讀取一個沒有偏移量或者偏移量無效的情況下（因為消費者長時間失效，包含偏移量的記錄已經過時並被刪除）該作何處理。默認是latest（消費者從最新的記錄開始讀取數據）。另一個值是             　      earliest（消費者從起始位置讀取partition的記錄）

6：enable.auto.commit，指定消費者是否自動提交偏移量，默認為true

7：partition.assignment.strategy，指定partition如何分配給消費者，默認是Range。Range：把Topic的若干個連續的partition分配給消費者。RoundRobin：把Topic的所有partition逐個分配給消費者

8：max.poll.records，單次調用poll方法能夠返回的消息數量

（5）提交和偏移量

1、消費者為什么要提交偏移量

當消費者崩潰或者有新的消費者加入，那么就會觸發再均衡（rebalance），完成再均衡后，每個消費者可能會分配到新的分區，而不是之前處理那個，為了能夠繼續之前的工作，消費者需要讀取每個partition最后一次提交的偏移量，然后從偏移量指定的地方繼續處理。

2、提交偏移量可能帶來的問題

case1：如果提交的偏移量小於客戶端處理的最后一個消息的偏移量，那么處於兩個偏移量之間的消息就會被重復處理。

case2：如果提交的偏移量大於客戶端處理的最后一個消息的偏移量，那么處於兩個偏移量之間的消息將會丟失。

 

 3、提交偏移量的方式

（1）自動提交 Automatic Commit

enable.auto.commit設置成true（默認為true），那么每過5s，消費者自動把從poll()方法接收到的最大的偏移量提交。提交的時間間隔由auto.commit.interval.ms控制，默認是5s

自動提交的優點是方便，但是可能會重復處理消息

（2）提交當前偏移量 Commit Current Offset

將enable.auto.commit設置成false，讓應用程序決定何時提交偏移量。commitSync()提交由poll()方法返回的最新偏移量，所以在處理完所有消息后要確保調用commitSync，否則會有消息丟失的風險。commitSync在提交成功或碰到無法恢復的錯誤之前，會一直重試。如果發生了再均衡，從最近一批消息到發生再均衡之間的所有消息都會被重復處理。

不足：broker在對提交請求作出回應之前，應用程序會一直阻塞，會限制應用程序的吞吐量

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("topic = %s, partition = %s, offset = %d,         
        customer = %s, country = %s\n", record.topic(),         
        record.partition(), record.offset(), record.key(), 
        record.value());
     }
    try {
        consumer.commitSync();//處理完當前批次的消息，在輪詢更多的消息之前，調用commitSync方法提交當前批次最新的消息
    } catch (CommitFailedException e) {
        log.error("commit failed", e);//只要沒有發生不可恢復的錯誤，commitSync方法會一直嘗試直至提交成功。如果提交失敗，我們也只能把異常記錄到錯誤日志里
    }
}        

（3）異步提交

異步提交的commitAsync，只管發送提交請求，無需等待broker響應。commitAsync提交之后不進行重試，假設要提交偏移量2000，這時候發生短暫的通信問題，服務器接收不到提交請求，因此也就不會作出響應。與此同時，我們處理了另外一批消息，並成功提交了偏移量3000,。如果commitAsync重新嘗試提交2000，那么它有可能在3000之后提交成功，這個時候如果發生再均衡，就會出現重復消息。

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("topic = %s, partition = %s, offset = %d,         
        customer = %s, country = %s\n", record.topic(),         
        record.partition(), record.offset(), record.key(), 
        record.value());
     }
     consumer.commitAsync(new OffsetCommitCallback() {//在broker作出響應后執行回調函數，回調經常被用於記錄提交錯誤或生成度量指標
　　  　　public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
　　　　　　if (e != null) {
　　　　　　　　log.error("Commit Failed for offsets {}", offsets, e);
　　　　　　}
　　　　　}});
}

（4）同步和異步組合提交

一般情況下，針對偶爾出現的提交失敗，不進行重試不會有太大的問題，因為如果提交失敗是因為臨時問題導致的，那么后續的提交總會有成功的。但是如果在關閉消費者或再均衡前的最后一次提交，就要確保提交成功。

因此，在消費者關閉之前一般會組合使用commitAsync和commitSync提交偏移量。

try {
　　while (true) {
   　　 ConsumerRecords<String, String> records = consumer.poll(100);
   　　 for (ConsumerRecord<String, String> record : records) {
   　　     System.out.println("topic = %s, partition = %s, offset = %d,         
     　　   customer = %s, country = %s\n", record.topic(),         
     　　   record.partition(), record.offset(), record.key(), 
      　　  record.value());
     　　}
       consumer.commitAsync();//如果一切正常，我們使用commitAsync來提交，這樣速度更快，而且即使這次提交失敗，下次提交很可能會成功
} catch (CommitFailedException e) {
    log.error("commit failed", e);
} finally {
    try {
　　　　consumer.commitSync();//關閉消費者前，使用commitSync，直到提交成成功或者發生無法恢復的錯誤
　　 } finally {
　　　　consumer.close();
　　 }
}

（5）提交特定的偏移量

消費者API允許調用commitSync()和commitAsync()方法時傳入希望提交的partition和offset的map，即提交特定的偏移量。

private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();//用於跟蹤偏移量的map int count = 0;
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("topic = %s, partition = %s, offset = %d,         
        customer = %s, country = %s\n", record.topic(),         
        record.partition(), record.offset(), record.key(), 
        record.value());//模擬對消息的處理
        //在讀取每條消息后，使用期望處理的下一個消息的偏移量更新map里的偏移量。下一次就從這里開始讀取消息
        currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1, “no matadata”));
        if (count++ % 1000 == 0) {//每處理1000條消息就提交一次偏移量，在實際應用中，可以根據時間或者消息的內容進行提交
            consumer.commitAsync(currentOffsets, null);
        }
    }

}  

（6）再均衡監聽器

在為消費者分配新的partition或者移除舊的partition時，可以通過消費者API執行一些應用程序代碼，在使用subscribe()方法時傳入一個ConsumerRebalanceListener實例。

ConsumerRebalanceListener需要實現的兩個方法

1：public void onPartitionRevoked(Collection<TopicPartition> partitions)方法會在再均衡開始之前和消費者停止讀取消息之后被調用。如果在這里提交偏移量，下一個接管partition的消費者就知道該從哪里開始讀取了。

2：public void onPartitionAssigned(Collection<TopicPartition> partitions)方法會在重新分配partition之后和消費者開始讀取消息之前被調用。

 

下面的例子演示如何在失去partition的所有權之前通過onPartitionRevoked()方法來提交偏移量。

private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();//用於跟蹤偏移量的map

      private class HandleRebalance implements ConsumerRebalanceListener {
        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        }

        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        //如果發生再均衡，要在即將失去partition所有權時提交偏移量。
        //注意：（1）提交的是最近處理過的偏移量，而不是批次中還在處理的最后一個偏移量。因為partition有可能在我們還在處理消息時被撤回。
        //(2)我們要提交所有分區的偏移量，而不只是即將市區所有權的分區的偏移量。因為提交的偏移量是已經處理過的，所以不會有什么問題。
        //(3)調用commitSync方法，確保在再均衡發生之前提交偏移量
            consumer.commitSync(currentOffsets);
        }
      }
      try{
          consumer.subscribe(topics, new HandleRebalance());
          while (true) {
              ConsumerRecords<String, String> records = consumer.poll(100);
              for (ConsumerRecord<String, String> record : records) {
                  System.out.println("topic = %s, partition = %s, offset = %d,         
                  customer = %s, country = %s\n", record.topic(),         
                  record.partition(), record.offset(), record.key(), 
                  record.value());//模擬對消息的處理
                  //在讀取每條消息后，使用期望處理的下一個消息的偏移量更新map里的偏移量。下一次就從這里開始讀取消息
                  currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1, “no matadata”));
              }
              consumer.commitAsync(currentOffsets, null);
      } catch(WakeupException e) {
          //忽略異常，正在關閉消費者
      } catch (Exception e) {
          log.error("unexpected error", e);
      } finally {
          try{
              consumer.commitSync(currentOffsets);
          } finally {
              consumer.close();
          }
      }

參考：《Kafka權威指南》