2. 程式人生 > >Kafaka的訊息消費方式

Kafaka的訊息消費方式

阿新 • • 發佈：2018-12-09

Kafaka的三種消費方式

1.消費位移確認

    Kafka消費者消費位移確認有自動提交與手動提交兩種策略。在建立KafkaConsumer物件時，通過引數enable.auto.commit設定，true表示自動提交（預設）。自動提交策略由消費者協調器（ConsumerCoordinator）每隔${auto.commit.interval.ms}毫秒執行一次偏移量的提交。手動提交需要由客戶端自己控制偏移量的提交。

        (1)自動提交。在建立一個消費者時，預設是自動提交偏移量，當然我們也可以顯示設定為自動。例如，我們建立一個消費者，該消費者自動提交偏移量

  1. Properties props =
    new Properties();
  2. props.put("bootstrap.servers", "localhost:9092");
  3. props.put("group.id", "test");
  4. props.put("client.id", "test");
  5. props.put("enable.auto.commit", true);// 顯示設定偏移量自動提交
  6. props.put("auto.commit.interval.ms", 1000);// 設定偏移量提交時間間隔
  7. props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"
    );
  8. props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  9. KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);// 建立消費者
  10. consumer.subscribe(Arrays.asList("test"));// 訂閱主題

        (2)手動提交。在有些場景我們可能對消費偏移量有更精確的管理，以保證訊息不被重複消費以及訊息不被丟失。假設我們對拉取到的訊息需要進行寫入資料庫處理，或者用於其他網路訪問請求等等複雜的業務處理，在這種場景下，所有的業務處理完成後才認為訊息被成功消費，這種場景下，我們必須手動控制偏移量的提交。

        Kafka 提供了非同步提交（commitAsync）及同步提交（commitSync）兩種手動提交的方式。兩者的主要區別在於同步模式下提交失敗時一直嘗試提交，直到遇到無法重試的情況下才會結束，同時，同步方式下消費者執行緒在拉取訊息時會被阻塞，直到偏移量提交操作成功或者在提交過程中發生錯誤。而非同步方式下消費者執行緒不會被阻塞，可能在提交偏移量操作的結果還未返

回時就開始進行下一次的拉取操作，在提交失敗時也不會嘗試提交。

        實現手動提交前需要在建立消費者時關閉自動提交，即設定enable.auto.commit=false。然後在業務處理成功後呼叫commitAsync()或commitSync()方法手動提交偏移量。由於同步提交會阻塞執行緒直到提交消費偏移量執行結果返回，而非同步提交併不會等消費偏移量提交成功後再繼續下一次拉取訊息的操作，因此非同步提交還提供了一個偏移量提交回調的方法commitAsync(OffsetCommitCallback callback)。當提交偏移量完成後會回撥OffsetCommitCallback 介面的onComplete()方法，這樣客戶端根據回撥結果執行不同的邏輯處理。

  1. Properties props = new Properties();
  2. props.put("bootstrap.servers", "localhost:9092");
  3. props.put("group.id", "test");
  4. props.put("client.id", "test");
  5. props.put("fetch.max.bytes", 1024);// 為了便於測試，這裡設定一次fetch 請求取得的資料最大值為1KB,預設是5MB
  6. props.put("enable.auto.commit", false);// 設定手動提交偏移量
  7. props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  8. props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  9. KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  10. // 訂閱主題
  11. consumer.subscribe(Arrays.asList("test"));
  12. try {
  13.     int minCommitSize = 10;// 最少處理10 條訊息後才進行提交
  14.     int icount = 0 ;// 訊息計算器
  15.     while (true) {
  16.         // 等待拉取訊息
  17.         ConsumerRecords<String, String> records = consumer.poll(1000);
  18.         for (ConsumerRecord<String, String> record : records) {
  19.             // 簡單打印出訊息內容,模擬業務處理
  20.             System.out.printf("partition = %d, offset = %d,key= %s value = %s%n", record. partition(), record.offset(), record.key(),record.value());
  21.             icount++;
  22.         }
  23.         // 在業務邏輯處理成功後提交偏移量
  24.         if (icount >= minCommitSize){
  25.             consumer.commitAsync(new OffsetCommitCallback() {
  26.                 @Override
  27.                 public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
  28.                     if (null == exception) {
  29.                     // TODO 表示偏移量成功提交
  30.                     System.out.println("提交成功");
  31.                     } else {
  32.                         // TODO 表示提交偏移量發生了異常，根據業務進行相關處理
  33.                         System.out.println("發生了異常");
  34.                     }
  35.                 }
  36.             });
  37.             icount=0; // 重置計數器
  38.         }
  39.     }
  40. } catch(Exception e){
  41.     // TODO 異常處理
  42.     e.printStackTrace();
  43. } finally {
  44.     consumer.close();
  45. }

2以時間戳查詢訊息

        Kafka 在0.10.1.1 版本增加了時間戳索引檔案，因此我們除了直接根據偏移量索引檔案查詢訊息之外，還可以根據時間戳來訪問訊息。consumer-API 提供了一個offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch)方法，該方法入參為一個Map 物件，Key 為待查詢的分割槽，Value 為待查詢的時間戳，該方法會返回時間戳大於等於待查詢時間的第一條訊息對應的偏移量和時間戳。需要注意的是，若待查詢的分割槽不存在，則該方法會被一直阻塞。

        假設我們希望從某個時間段開始消費，那們就可以用offsetsForTimes()方法定位到離這個時間最近的第一條訊息的偏移量，在查到偏移量之後呼叫seek(TopicPartition partition, long offset)方法將消費偏移量重置到所查詢的偏移量位置，然後呼叫poll()方法長輪詢拉取訊息。例如，我們希望從主題“stock-quotation”第0 分割槽距離當前時間相差12 小時之前的位置開始拉取訊息

  1. Properties props = new Properties();
  2. props.put("bootstrap.servers", "localhost:9092");
  3. props.put("group.id", "test");
  4. props.put("client.id", "test");
  5. props.put("enable.auto.commit", true);// 顯示設定偏移量自動提交
  6. props.put("auto.commit.interval.ms", 1000);// 設定偏移量提交時間間隔
  7. props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  8. props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  9. KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  10. // 訂閱主題
  11. consumer.assign(Arrays.asList(new TopicPartition("test", 0)));
  12. try {
  13.     Map<TopicPartition, Long> timestampsToSearch = new HashMap<TopicPartition,Long>();
  14.     // 構造待查詢的分割槽
  15.     TopicPartition partition = new TopicPartition("stock-quotation", 0);
  16.     // 設定查詢12 小時之前訊息的偏移量
  17.     timestampsToSearch.put(partition, (System.currentTimeMillis() - 12 * 3600 * 1000));
  18.     // 會返回時間大於等於查詢時間的第一個偏移量
  19.     Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes (timestampsToSearch);
  20.     OffsetAndTimestamp offsetTimestamp = null;
  21.     // 這裡依然用for 輪詢，當然由於本例是查詢的一個分割槽，因此也可以用if 處理
  22.     for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsetMap.entrySet()) {
  23.         // 若查詢時間大於時間戳索引檔案中最大記錄索引時間，
  24.         // 此時value 為空,即待查詢時間點之後沒有新訊息生成
  25.         offsetTimestamp = entry.getValue();
  26.         if (null != offsetTimestamp) {
  27.         // 重置消費起始偏移量
  28.         consumer.seek(partition, entry.getValue().offset());
  29.         }
  30.     }
  31.     while (true) {
  32.         // 等待拉取訊息
  33.         ConsumerRecords<String, String> records = consumer.poll(1000);
  34.         for (ConsumerRecord<String, String> record : records){
  35.             // 簡單打印出訊息內容
  36.             System.out.printf("partition = %d, offset = %d,key= %s value = %s%n", record.partition(), record.offset(), record.key(),record.value());
  37.         }
  38.     }
  39. } catch (Exception e) {
  40.     e.printStackTrace();
  41. } finally {
  42.     consumer.close();
  43. }

3消費速度控制

        提供 pause(Collection<TopicPartition> partitions)和resume(Collection<TopicPartition>
partitions)方法，分別用來暫停某些分割槽在拉取操作時返回資料給客戶端和恢復某些分割槽向客戶端返回資料操作。通過這兩個方法可以對消費速度加以控制，結合業務使用。

Kafaka的三種消費方式

1.消費位移確認

    Kafka消費者消費位移確認有自動提交與手動提交兩種策略。在建立KafkaConsumer物件時，通過引數enable.auto.commit設定，true表示自動提交（預設）。自動提交策略由消費者協調器（ConsumerCoordinator）每隔${auto.commit.interval.ms}毫秒執行一次偏移量的提交。手動提交需要由客戶端自己控制偏移量的提交。

        (1)自動提交。在建立一個消費者時，預設是自動提交偏移量，當然我們也可以顯示設定為自動。例如，我們建立一個消費者，該消費者自動提交偏移量

  1. Properties props = new Properties();
  2. props.put("bootstrap.servers", "localhost:9092");
  3. props.put("group.id", "test");
  4. props.put("client.id", "test");
  5. props.put("enable.auto.commit", true);// 顯示設定偏移量自動提交
  6. props.put("auto.commit.interval.ms", 1000);// 設定偏移量提交時間間隔
  7. props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  8. props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  9. KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);// 建立消費者
  10. consumer.subscribe(Arrays.asList("test"));// 訂閱主題

        (2)手動提交。在有些場景我們可能對消費偏移量有更精確的管理，以保證訊息不被重複消費以及訊息不被丟失。假設我們對拉取到的訊息需要進行寫入資料庫處理，或者用於其他網路訪問請求等等複雜的業務處理，在這種場景下，所有的業務處理完成後才認為訊息被成功消費，這種場景下，我們必須手動控制偏移量的提交。

        Kafka 提供了非同步提交（commitAsync）及同步提交（commitSync）兩種手動提交的方式。兩者的主要區別在於同步模式下提交失敗時一直嘗試提交，直到遇到無法重試的情況下才會結束，同時，同步方式下消費者執行緒在拉取訊息時會被阻塞，直到偏移量提交操作成功或者在提交過程中發生錯誤。而非同步方式下消費者執行緒不會被阻塞，可能在提交偏移量操作的結果還未返

回時就開始進行下一次的拉取操作，在提交失敗時也不會嘗試提交。

        實現手動提交前需要在建立消費者時關閉自動提交，即設定enable.auto.commit=false。然後在業務處理成功後呼叫commitAsync()或commitSync()方法手動提交偏移量。由於同步提交會阻塞執行緒直到提交消費偏移量執行結果返回，而非同步提交併不會等消費偏移量提交成功後再繼續下一次拉取訊息的操作，因此非同步提交還提供了一個偏移量提交回調的方法commitAsync(OffsetCommitCallback callback)。當提交偏移量完成後會回撥OffsetCommitCallback 介面的onComplete()方法，這樣客戶端根據回撥結果執行不同的邏輯處理。

  1. Properties props = new Properties();
  2. props.put("bootstrap.servers", "localhost:9092");
  3. props.put("group.id", "test");
  4. props.put("client.id", "test");
  5. props.put("fetch.max.bytes", 1024);// 為了便於測試，這裡設定一次fetch 請求取得的資料最大值為1KB,預設是5MB
  6. props.put("enable.auto.commit", false);// 設定手動提交偏移量
  7. props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  8. props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  9. KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  10. // 訂閱主題
  11. consumer.subscribe(Arrays.asList("test"));
  12. try {
  13.     int minCommitSize = 10;// 最少處理10 條訊息後才進行提交
  14.     int icount = 0 ;// 訊息計算器
  15.     while (true) {
  16.         // 等待拉取訊息
  17.         ConsumerRecords<String, String> records = consumer.poll(1000);
  18.         for (ConsumerRecord<String, String> record : records) {
  19.             // 簡單打印出訊息內容,模擬業務處理
  20.             System.out.printf("partition = %d, offset = %d,key= %s value = %s%n", record. partition(), record.offset(), record.key(),record.value());
  21.             icount++;
  22.         }
  23.         // 在業務邏輯處理成功後提交偏移量
  24.         if (icount >= minCommitSize){
  25.             consumer.commitAsync(new OffsetCommitCallback() {
  26.                 @Override
  27.                 public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
  28.                     if (null == exception) {
  29.                     // TODO 表示偏移量成功提交
  30.                     System.out.println("提交成功");
  31.                     } else {
  32.                         // TODO 表示提交偏移量發生了異常，根據業務進行相關處理
  33.                         System.out.println("發生了異常");
  34.                     }
  35.                 }
  36.             });
  37.             icount=0; // 重置計數器
  38.         }
  39.     }
  40. } catch(Exception e){
  41.     // TODO 異常處理
  42.     e.printStackTrace();
  43. } finally {
  44.     consumer.close();
  45. }

2以時間戳查詢訊息

        Kafka 在0.10.1.1 版本增加了時間戳索引檔案，因此我們除了直接根據偏移量索引檔案查詢訊息之外，還可以根據時間戳來訪問訊息。consumer-API 提供了一個offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch)方法，該方法入參為一個Map 物件，Key 為待查詢的分割槽，Value 為待查詢的時間戳，該方法會返回時間戳大於等於待查詢時間的第一條訊息對應的偏移量和時間戳。需要注意的是，若待查詢的分割槽不存在，則該方法會被一直阻塞。

        假設我們希望從某個時間段開始消費，那們就可以用offsetsForTimes()方法定位到離這個時間最近的第一條訊息的偏移量，在查到偏移量之後呼叫seek(TopicPartition partition, long offset)方法將消費偏移量重置到所查詢的偏移量位置，然後呼叫poll()方法長輪詢拉取訊息。例如，我們希望從主題“stock-quotation”第0 分割槽距離當前時間相差12 小時之前的位置開始拉取訊息

  1. Properties props = new Properties();
  2. props.put("bootstrap.servers", "localhost:9092");
  3. props.put("group.id", "test");
  4. props.put("client.id", "test");
  5. props.put("enable.auto.commit", true);// 顯示設定偏移量自動提交
  6. props.put("auto.commit.interval.ms", 1000);// 設定偏移量提交時間間隔
  7. props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  8. props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  9. KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  10. // 訂閱主題
  11. consumer.assign(Arrays.asList(new TopicPartition("test", 0)));
  12. try {
  13.     Map<TopicPartition, Long> timestampsToSearch = new HashMap<TopicPartition,Long>();
  14.     // 構造待查詢的分割槽
  15.     TopicPartition partition = new TopicPartition("stock-quotation", 0);
  16.     // 設定查詢12 小時之前訊息的偏移量
  17.     timestampsToSearch.put(partition, (System.currentTimeMillis() - 12 * 3600 * 1000));
  18.     // 會返回時間大於等於查詢時間的第一個偏移量
  19.     Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes (timestampsToSearch);
  20.     OffsetAndTimestamp offsetTimestamp = null;
  21.     // 這裡依然用for 輪詢，當然由於本例是查詢的一個分割槽，因此也可以用if 處理
  22.     for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsetMap.entrySet()) {
  23.         // 若查詢時間大於時間戳索引檔案中最大記錄索引時間，
  24.         // 此時value 為空,即待查詢時間點之後沒有新訊息生成
  25.         offsetTimestamp = entry.getValue();
  26.         if (null != offsetTimestamp) {
  27.         // 重置消費起始偏移量
  28.         consumer.seek(partition, entry.getValue().offset());
  29.         }
  30.     }
  31.     while (true) {
  32.         // 等待拉取訊息
  33.         ConsumerRecords<String, String> records = consumer.poll(1000);
  34.         for (ConsumerRecord<String, String> record : records){
  35.             // 簡單打印出訊息內容
  36.             System.out.printf("partition = %d, offset = %d,key= %s value = %s%n", record.partition(), record.offset(), record.key(),record.value());
  37.         }
  38.     }
  39. } catch (Exception e) {
  40.     e.printStackTrace();
  41. } finally {
  42.     consumer.close();
  43. }

3消費速度控制

        提供 pause(Collection<TopicPartition> partitions)和resume(Collection<TopicPartition>
partitions)方法，分別用來暫停某些分割槽在拉取操作時返回資料給客戶端和恢復某些分割槽向客戶端返回資料操作。通過這兩個方法可以對消費速度加以控制，結合業務使用。

相關推薦

Kafaka的訊息消費方式

Kafaka的三種消費方式 1.消費位移確認     Kafka消費者消費位移確認有自動提交與手動提交兩種策略。在建立KafkaConsumer物件時，通過引數enable.auto.commit設

RocketMQ專題2：三種常用生產消費方式(順序、廣播、定時)以及順序消費源碼探究

channel quest 再次 指定 conf 默認 廣播 body tag 順序、廣播、定時任務 前插 ? 在進行常用的三種消息類型例子展示的時候，我們先來說一說RocketMQ的幾個重要概念： PullConsumer與PushConsumer：主要區別在於Pul

CloudFoundry命令行和Kubernetes命令行的Restful API消費方式

ner err 修改 path include ssh rec 一個 expires 先說CloudFoundry的命令行工具CLI。我們在CloudFoundry環境下工作，第一個使用的命令就是cf login。 如果在環境變量裏維護CF_TRACE的值為true： 則

hadoop（十四）kafaka訊息佇列

kafka筆記  1/kafka是一個分散式的訊息快取系統 2/kafka叢集中的伺服器都叫做broker 3/kafka有兩類客戶端，一類叫producer（訊息生產者），一類叫做consumer（訊息消費者），客戶端和broker伺服器之間採用tcp協議連線 4/kafka

DefaultMessageListenerContainer訊息應答方式

JMS標準支援的訊息應答模式有下面幾種： AUTO_ACKNOWLEDGE：自動應答，預設的應答方式 CLIENT_ACKNOWLEDGE：客戶端應答，應答由應用程式在接收到訊息後觸發 DUPS_OK_ACKNOWLEDGE：儘量不要使用，如果使用這種方式，應用程式需要考慮處理訊息重

RabbitMQ消費方式彙總

在學習本章節前，請先學習之前的章節：Java訪問RabbitMQ：https://www.cnblogs.com/duanjt/p/10057330.htmlRabbitMQ訊息釋出時的權衡：https://www.cnblogs.com/duanjt/p/10075308.html 一、推送Consume

RabbitMQ消費方式匯總

都是 efault esp 實現 取模 != 之前 handle tin 在學習本章節前，請先學習之前的章節：Java訪問RabbitMQ：https://www.cnblogs.com/duanjt/p/10057330.htmlRabbitMQ消息發布時的權衡：http

Kafka中Broker儲存訊息的方式

1.儲存方式 物理上把topic分成一個或多個patition(對應 server.properties 中的num.partitions=3配置)，每個patition物理上對應一個檔案 (該資料夾儲存該patition的所有訊息和索引檔案) 2.儲存策略 無論訊息

RocketMQ專題2：三種常用生產消費方式(順序、廣播、定時)以及順序消費原始碼探究

順序、廣播、定時任務 前插 ​ 在進行常用的三種訊息型別例子展示的時候，我們先來說一說RocketMQ的幾個重要概念： PullConsumer與PushConsumer：主要區別在於Pull與Push的區別。對於PullConsumer，消費者會主動從broker中拉取訊息進行消費。而對於PushCon

Spring Cloud Stream + RabbitMQ 訊息生成和訊息消費

在本 DEMO中有兩個節點互為訊息的生產者和訊息消費者。 一、節點1 1. pom.xml <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/P

RocketMQ的訊息傳送方式

同步傳送 簡單來說，同步傳送就是指 producer 傳送訊息後，會在接收到 broker 響應後才繼續發下一條訊息的通訊方式。 由於這種同步傳送的方式確保了訊息的可靠性，同時也能及時得到訊息傳送的結果，故而適合一些傳送比較重要的訊息場景，比如說重要的通知郵件、營銷簡訊等等。在實

RocketMQ 訊息消費

訊息消費 難點：如何保證訊息只消費一次？   消費模式： 　　1、單一消費模式：一條訊息，僅被一個消費者進行消費。 　　　　如何進行負載？負載演算法有 a、平均分配。b、平均輪詢分配。c、一致性hash（不推薦）。d、根據配置，為消費者指定訊息佇列。e、根據broker進行分配，消費者固定

(四) RabbitMQ實戰教程(面向Java開發人員)之@RabbitListener訊息消費

使用RabbitListener註解進行訊息消費 在前一篇部落格中我們往MessageListenerContainer設定了MessageListener進行訊息的消費，本篇部落格將介紹一種更為簡單的訊息消費方式:使用@RabbitListener註解方式。

RocketMQ——Consumer篇：PULL模式下的訊息消費（DefaultMQPullConsumer）

1 應用層的使用方式 在應用層初始化DefaultMQPullConsumer類，然後呼叫該類的start方法啟動Consumer；接下來的消費步驟如下： 1、呼叫DefaultMQPullConsumer.fetchSubscribeMessageQueu

基於java的nio訊息實現方式優缺點分析及示例程式碼說明

轉：http://www.xwood.net/_site_domain_/_root/5870/5874/t_c255146.html java nio在jdk1.4版本中加入新io的功能，所有涉及類都在java.nio包下，改寫了原有客戶端伺服器一對一執行緒阻塞等待的方式

Kafka入門三：幾種消費方式

1.消費位移確認    Kafka消費者消費位移確認有自動提交與手動提交兩種策略。在建立KafkaConsumer物件時，通過引數enable.auto.commit設定，true表示自動提交（預設）。自動提交策略由消費者協調器（ConsumerCoordinator）每隔$

Spring Cloud Alibaba基礎教程：支持的幾種服務消費方式（RestTemplate、WebClient、Feign）

功能 code vat learn master public http stp spa 通過《Spring Cloud Alibaba基礎教程：使用Nacos實現服務註冊與發現》一文的學習，我們已經學會如何使用Nacos來實現服務的註冊與發現，同時也介紹如何通過LoadB

原始碼分析RocketMQ之訊息消費

1、訊息消費關注點 1）訊息消費方式：拉取、推送 2）消費者組與消費模式 consumerGroup;       MessageModel messageModel;    多個消費者組成一個消費組，兩種模式：叢集（訊息被其中任何一個訊息者消費）、廣播模式（全部消費者消費

Fetcher: KafkaConsumer訊息消費的管理者

我們在客戶端使用KafkaConsumer類進行Kafka訊息的消費，其實KafkaConsumer是將建立消費請求、接收響應的操作全部交給了Fetcher去處理。我們從KafkaConsumer.poll()方法進入，解析Fetcher的工作流程。 在我們看

kafka的topic多分割槽的情況，如何保證跨區的訊息消費的順序性

這個問題嚴格來說是肯定有的，kafka只能保證分割槽內的有序性。 下面是kafka作者Jay Kreps的blog中介紹kafka設計思想的一段話。 Each partition is a totally ordered log, but there is no global ordering betwe