取用訊息

配置消費者內容

消費者有許多配置設定，可控制其行為的各個層面。 下列設定是一些最重要的設定。

還有許多其他配置設定可用，但在開始實驗之前，請先閱讀 Apache Kafka 說明文件。

消耗者群組

_消費者組_是一組合作消費來自一個或多個主題的訊息的消費者。 群組中的消費者都使用相同的 group.id 配置值。 如果您需要有多個消費者來處理工作負載，則可以在同一個消費者群組中執行多個消費者。 即使您只需要一個使用者，通常也會為 group.id 指定一個值。

每個消費者群組在叢集中都有一個伺服器，稱為_協調器，_負責為群組中的消費者分配分區。 此責任分佈在叢集裡的伺服器上，甚至是負載。 每次群組重新平衡時，都可以變更消費者的分割區指派。

當消費者結合消費者群組時，它會探索群組的協調者。 然後，消費者告訴協調器它想要加入該組，協調器開始重新平衡該組中的分區以包含新成員。

單一分割區中的訊息只由每一個群組中的一個消費者處理。 這種機制保證了每個分區上的消息都按順序處理。 請參閱下圖，以取得一個範例，其中主題包含三個分割區，而耗用該主題的消費者群組包含兩個消費者。 群組中的一個消費者獲指派兩個分割區，而另一個消費者則獲指派一個分割區。

當消費者群組中發生以下變更之一時，該群組會透過將分割區分配轉移給群組成員來重新平衡以適應變更：

• 消費者加入該組。
• 消費者離開該組。
• 協調者認為消費者不再存在。
• 新分區將會新增到現有主題中。

如果您有一個重新平衡的消費者群組，請注意，任何離開該群組的消費者的提交都會被拒絕，直到它重新加入該群組。 在此情況下，消費者需要重新加入群組，在群組中它可能獲指派一個不同於先前所取用的分割區。

消費者活動

Kafka 會自動偵測失敗的消費者，以將分割區重新指派給工作中消費者。 它使用兩種機制: 輪詢和活動訊號。

如果從 Consumer.poll(...) 返回的訊息批次很大或處理很耗時，則再次呼叫 poll() 之前的延遲可能會很大或不可預測。 在某些情況下，需要配置較長的最大輪詢間隔，這樣消費者就不會因為訊息處理需要較長的時間而被從其群組移除。 如果此機制是唯一可用的機制，則偵測失敗消費者所花費的時間也會很長。

為了讓消費者活動更容易處理，Kafka 0.10.1 新增了背景活動訊號。 群組協調者期望群組成員會定期傳送活動訊號，以表示它們仍處於活動狀態。 後台心跳線程在消費者中運行，並向協調器發送定期心跳。 如果協調器在_會話逾時_內沒有收到來自組成員的心跳，則協調器將從群組中刪除該成員並開始群組的重新平衡。 會話逾時可以比最大輪詢間隔短得多，這樣即使訊息處理需要很長時間，檢測失敗的消費者所需的時間也可以很短。

您可以使用 max.poll.interval.ms 屬性配置最大輪詢間隔，並使用 session.timeout.ms 屬性配置會話逾時。 除非處理一批訊息需要超過 5 分鐘，否則不需要使用這些設定。

管理偏移

對於每個消費者群組，Kafka維護每個消費分區的提交偏移量。 當消費者處理訊息時，並不會將訊息從分割區移除。 相反，它只是透過使用稱為提交偏移量的過程來更新其當前偏移量。

Event Streams 會將已確定的偏移資訊保留 7 天。

控制訊息取用速度

如果您遇到因訊息氾濫而導致的訊息處理問題，您可以設定消費者選項來控制訊息消費的速度。 使用 fetch.max.bytes 及 max.poll.records 來控制呼叫 poll() 可傳回的資料量。

處理消費者重新平衡

當消費者加入群組或從群組中刪除時，會發生群組重新平衡，且消費者無法消費訊息。 這會導致消費者組中的所有消費者在短時間內無法使用。

如果您收到「分區撤銷」回呼通知，請使用ConsumerRebalanceListener手動提交偏移量（如果您不使用自動提交）並暫停進一步處理，直到使用「分區分配」回呼通知您成功重新平衡。

程式碼 Snippet

這些程式碼片段在較高層次上說明了所涉及的概念。 如需完整範例，請參閱 GitHub中的 Event Streams 範例。

若要將消費者連接至 Event Streams，您需要建立服務認證。 如需如何取得這些認證的相關資訊，請參閱 連接至 Event Streams。

在消費者程式碼中，您首先需要建置一組配置內容。 與Event Streams的所有連線均透過使用 TLS 和使用者密碼驗證進行保護，因此您至少需要這些屬性。 將 BOOTSTRAP_ENDPOINTS、USER 和 PASSWORD 替換為您自己的服務憑證中的內容：

Properties props = new Properties();
 props.put("bootstrap.servers", BOOTSTRAP_ENDPOINTS);
 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USER\" password=\"PASSWORD\";");
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "PLAIN");
 props.put("ssl.protocol", "TLSv1.2");
 props.put("ssl.enabled.protocols", "TLSv1.2");
 props.put("ssl.endpoint.identification.algorithm", "HTTPS");

若要耗用訊息，您還需要指定索引鍵及值的解除序列化程式，如下列範例所示。

 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

這些解除序列化程式必須符合生產者使用的序列化程式。

然後，使用 KafkaConsumer 來取用訊息，每一則訊息都由一個 ConsumerRecord 表示。 取用訊息最常見的方式是藉由設定群組 ID 來將消費者放入消費者群組中，然後呼叫 subscribe() 以取得主題清單。 消費者被分配了一些要消費的分區，但如果組中存在的消費者多於主題中的分區，則可能不會為消費者分配任何分區。 接下來，請在迴圈中呼叫 poll()，接收要處理的訊息批次，每一則訊息都由一個 ConsumerRecord 表示。

props.put("group.id", "G1");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));  // T1 is the topic name
while (true) {
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
   for (ConsumerRecord<String, String> record : records)
     System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

這個消費者循環永遠運行，但可以透過呼叫 Consumer.wakeup() 來從另一個線程中斷它，以實現整齊的關閉。

若要手動確定偏移，必須先將 enable.auto.commit 配置設為 false。 然後，使用 Consumer.commmitSync() 或 Consumer.commitAsync()，定期更新消費者的已確定偏移。 為了簡單起見，此範例處理每一個分割區的記錄並分別確定最後的偏移。

props.put("group.id", "G1");
props.put("enable.auto.commit", "false");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> partRecords = records.records(tp);
      long lastOffset = 0;
      for (ConsumerRecord<String, String> record : partRecords) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        lastOffset = record.offset();
      }
      // having processed all the records in the above loop, we commit the partition's offset to 1 more than the last offset
      consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastOffset + 1)));
    }
  }
}
finally {
  consumer.close();
}

異常狀況處理

任何使用 Kafka 用戶端的健全應用程式都需要處理某些預期狀況下的異常狀況。 在某些情況下，不會直接引發異常，因為某些方法是非同步的，並透過使用 Future 或回調來傳遞其結果。 請參閱 GitHub 中顯示完整範例的範例程式碼。

處理程式碼中的下列異常狀況清單: