知っておくべき Kafka コンシューマーのパラメーターは何ですか?

[[407801]]

この記事では、Kafka Consumer について簡単に紹介します。これは、Kafka Conumer を詳細に学習するための窓口となります。それは主に次の3つの側面から始まります。

• コアパラメータ
• コアコンポーネント
• コアAPI

1. Kafka コンシューマーコアパラメータの概要

個人的には、Kafka Consumer のコア動作メカニズムを深く理解したい場合は、コアパラメータから始めて、キューロードメカニズム、メッセージプルモデル、消費モデル、サイト送信などのメカニズムをより深く理解するための基礎を築くことができると思います。

Kafka Consumer のコア プロパティは ConsumerConfig で定義されます。

1.1 基本関数パラメータ

• グループID

コンシューマー グループ名。

• クライアントID

クライアント識別 ID。デフォルトはコンシューマー シーケンス番号です。実際には、コンシューマー グループ内で繰り返すことができないクライアント IP アドレスを含めることをお勧めします。

• ブートストラップサーバー

ブローカー サーバーのアドレス リスト。

• クライアント.dns.ルックアップ

クライアントは、ブートストラップ アドレスを見つけるために次の 2 つの方法をサポートしています。

• 解決正規ブートストラップサーバーのみ

このメソッドは、ホスト上のネーム サービスに従って bootstrap.servers によって提供されるホスト名 (hostname) に基づいて IP アドレスの配列 (InetAddress.getAllByName) を返し、次に inetAddress.getCanonicalHostName() を取得して、TCP 接続を確立します。

ホストは複数のネットワーク カードで構成できます。この機能を有効にすると、複数のネットワーク カードの利点を有効に活用し、ブローカーのネットワーク負荷を軽減できるようになります。

• すべてのDNSIPを使用する

このメソッドは、bootstrap.servers で提供されるホスト名とポートを直接使用して TCP 接続を作成します。これはデフォルトのオプションです。

• 自動コミットを有効にする

自動サイト送信を有効にするかどうか。デフォルトは true です。

• 自動コミット間隔 ms

自動サイト送信が有効になっている場合、サイト送信頻度はデフォルトで 5 秒に設定されます。

• パーティション割り当て戦略

コンシューマ キューの負荷アルゴリズム。デフォルトは間隔による平均分散 (RangeAssignor)、オプションの値: ポーリング (RoundRobinAssignor)

• 自動オフセットリセット

サイト戦略をリセットします。 Kafka がサイトを送信すると、対応するメッセージが削除されます。デフォルトは最新です。オプションは earliest と none (例外がスローされます) です。

• キーデシリアライザー

使用されるキーシリアル化クラス

• 値.デシリアライザー

メッセージ本文のシリアル化クラス

• インターセプタークラス

コンシューマー側インターセプターは複数存在する場合があります。

• チェック.crcs

CRC を消費者側でチェックする必要があるかどうか。デフォルトは true です。

1.2 ネットワーク関連パラメータ

• 送信バッファバイト

ネットワーク チャネル (TCP) の送信バッファ サイズはデフォルトで 128K です。

• 受信バッファバイト

ネットワーク チャネル (TCP) の受信バッファ サイズはデフォルトで 32K です。

• 再接続.backoff.ms

リンクを再確立するためのデフォルトの待機時間は 50 ミリ秒です。これは低レベルのネットワーク パラメータであり、注意する必要はありません。

• 再接続バックオフ最大ミリ秒

接続を再確立するための最大待機時間は、デフォルトでは 1 秒です。同じ接続に 2 回連続して再接続すると、reconnect.backoff.ms の初期値に基づいて待機時間が指数関数的に増加します。ただし、最大値を超えると、指数関数的に増加しなくなります。

• 再試行.バックオフ.ms

再試行間隔。デフォルト値は 100 ミリ秒です。

• 接続最大アイドル時間

接続の最大アイドル時間。デフォルトは 9 秒です。

• リクエストタイムアウト ミリ秒

リクエストタイムアウト、ブローカーとのネットワーク通信のリクエストタイムアウト。

1.3 コア動作パラメータ

• 最大投票レコード数

ポーリング メソッド呼び出しごとに取得できるメッセージの最大数は、デフォルトでは 500 です。

• 最大ポーリング間隔（ミリ秒）

2 つのポーリング メソッド呼び出し間の最大間隔 (ミリ秒単位)。デフォルトは 5 分です。コンシューマーが間隔内にポーリング操作を開始しない場合、コンシューマーは削除され、再バランスがトリガーされ、コンシューマーに割り当てられたキューは他のコンシューマーに割り当てられます。

• セッションタイムアウト ms

コンシューマーとブローカー間のハートビート タイムアウトは、デフォルトでは 10 秒です。ブローカーが指定された時間内にハートビート要求を受信しない場合、ブローカーはコンシューマーを削除し、リバランスをトリガーします。

• ハートビート間隔.ms

ハートビート間隔。コンシューマーはこの頻度でブローカーにハートビートを送信します。デフォルトは 3 秒です。これは主にセッションが期限切れにならないようにするためです。

• フェッチ最小バイト数

メッセージをプルするときに返される最小バイト数。デフォルト値は 1 バイトです。

• フェッチ最大バイト数

一度にメッセージをプルするときに返される最大バイト数。デフォルト値は 1M です。パーティション内の最初のメッセージ バッチのサイズがこの値より大きい場合は、それらも返されます。

• 最大パーティションフェッチバイト数

各パーティションから一度に取得できる最大バイト数は、デフォルトでは 1M です。

• フェッチ最大待ち時間（ミリ秒）

データが fetch.min.bytes を満たすまでフェッチが待機する最大待機時間。

• メタデータの最大年齢

クライアント上のメタデータの有効期限。有効期限が切れると、クライアントはブローカーから最新のメタデータを再度取得します。デフォルトは 5 分です。

• 内部グループを終了時に退会する

消費者が契約終了後すぐにサブスクリプション グループを離れるかどうか。デフォルト値は true です。これは、クライアントが切断されるとすぐに再バランス調整がトリガーされることを意味します。 false に設定すると、再バランスはすぐにはトリガーされず、セッションの有効期限が切れた後にトリガーされます。

2. KafkaConsumer コアコンポーネントと API

KafkaConsumer のコア パラメータを通じて、基本的に Kafka のコア ポイントを垣間見ることができます。次に、Kafka コンシューマー消費モデルに関するさらなる詳細な調査の基礎を築くために、KafkaConsumer のコア コンポーネントを紹介します。

2.1 コアコンポーネント

KafkaConsumer は次のコア コンポーネントで構成されています。

• コンシューマーネットワーククライアント

コンシューマー側のネットワーク クライアントは、基盤となるネットワーク通信を提供し、クライアントとサーバー間の RPC 通信を担当します。

• 消費者コーディネーター

消費者側コーディネーター。 Kafka の設計では、クラスター内の各コンシューマー グループは、コンシューマー グループのコーディネーターとなるブローカー ノードを選出します。ブローカー ノードは、コンシューマー グループの状態管理、特にコンシューマー グループの再調整 (コンシューマーの参加と退出) を担当します。このタイプのやり取りは、消費者とブローカー コーディネーターの間で行われます。

• フェッチャー

メッセージプル。

ヒント: この記事では、各コンポーネントを詳しく説明するつもりはありません。この記事の前半で各パラメータの意味を理解した上で、これらのパラメータが最終的にどのコンポーネントに履歴書が渡されるかを比較し、詳細に検討することをお勧めします。

2.2 コアAPIの概要

最後に、コンシューマーのコア API を見てみましょう。

• <トピックパーティション> 割り当てを設定する()

このコンシューマーのキュー割り当てのリストを取得します。

• Set<String> サブスクリプション()

このコンシューマーのサブスクリプション情報を取得します。

• void subscribe(コレクション<String> トピック)

トピックを購読します。

• void subscribe(Collection<String> トピック、ConsumerRebalanceListener コールバック)

トピックをサブスクライブし、キューの再バランス調整のリスナーを指定します。

• void 割り当て(コレクション<TopicPartition> パーティション)

サブスクリプションの代わりに、消費するキューを手動で指定します。

• void 購読解除()

サブスクリプションを解除します。

• 消費者記録

ポーリング(期間タイムアウト)

メッセージのプルは KafkaConsumer のコアメソッドであり、以下で詳しく説明します。

• void commitSync()

この消費送信バッチの消費進行状況を同期的に送信します。これについては、以降の記事で詳しく説明します。

• void commitSync(期間タイムアウト)

消費の進行状況を同期的に送信し、タイムアウトを設定します。

• void commitSync(マップ

オフセット)

消費の同期送信の進行状況を表示します。 offsets は、送信する必要がある消費進捗の情報を示します。

• void commitSync(最終マップ

オフセット、最終期間タイムアウト)

タイムアウト期間を含む、同期送信の消費の進行状況を表示します。

• void seek(トピックパーティション パーティション、long オフセット)

consumer#poll メソッドによる次のメッセージ プルのオフセットをリセットします。

• void seek(トピックパーティション パーティション、オフセットとメタデータ オフセットとメタデータ)

seek メソッドはメソッドをオーバーロードします。

• void seekToBeginning(コレクション<TopicPartition> パーティション)

ポーリング メソッドの次のプル オフセットをキューの初期オフセットに設定します。

• void seekToEnd(コレクション<TopicPartition> パーティション)

ポーリング メソッドの次のプル オフセットをキューの最大オフセットに設定します。

• ロングポジション(トピックパーティションパーティション)

プルされるオフセットを取得します。

• ロングポジション(TopicPartition パーティション、最終期間タイムアウト)

同上。

• OffsetAndMetadata がコミットされました (TopicPartition パーティション)

指定されたパーティションのコミットされたオフセットを取得します。

• OffsetAndMetadata がコミットされました (TopicPartition パーティション、最終期間タイムアウト)

同上。

• マップメトリクス()

統計指標。

• List<PartitionInfo> パーティション(文字列トピック)

トピックのルーティング情報を取得します。

• List<PartitionInfo> パーティションFor(文字列トピック、期間タイムアウト)

同上。

• マップリストトピック()

すべてのトピックのルーティング情報を取得します。

• マップリストトピック(期間タイムアウト)

同上。

• Set< TopicPartition> paused()

中断されたパーティション情報を取得します。

• void pause(コレクション<TopicPartition> パーティション)

パーティションを一時停止すると、次のポーリング メソッドではこれらのパーティションのメッセージが返されなくなります。

• void resume(Collection<TopicPartition> パーティション)

一時停止されたパーティションを再開します。

• 地図

オフセットForTimes(マップタイムスタンプを検索)

タイムスタンプに基づいて最新のメッセージのオフセットを見つけます。

• 地図

offsetsForTimes(MaptimestampsToSearch、期間タイムアウト)

同上。

• 地図

beginningOffsets(Collection<TopicPartition> パーティション)

指定されたパーティションの現在の最小オフセットを照会します。

• 地図

beginningOffsets(Collection<TopicPartition> パーティション、期間タイムアウト)

同上。

• 地図

endOffsets(Collection<TopicPartition> パーティション)

指定されたパーティションの現在の最大オフセットを照会します。

• 地図

endOffsets(Collection<TopicPartition> パーティション、期間タイムアウト)

同上。

• void close()

コンシューマーを閉じます。

• void close(期間タイムアウト)

コンシューマーを閉じます。

• void ウェイクアップ()

消費者を目覚めさせましょう。

Kafka が提供するコンシューマーは、メッセージを自動的にプルするプッシュ モードを提供する RocketMQ とは異なります。アプリケーションは、これらの API を自動的に整理してメッセージをプルする必要があります。

Kafka コンシューマーもサイトの自動送信メカニズムをサポートしていることは注目に値します。 Kafka のコンシューマー (KafkaConsumer) オブジェクトはスレッドセーフではありません。

KafkaConsumer の一時停止 (特定のパーティションの消費を一時停止) と再開 (特定のパーティションの消費を再開) に基づいて、コンシューマー側の電流制限メカニズムを簡単に実装できます。

この記事は主に消費者についての一般的な理解を深めることを目的としています。次回以降も、消費者のコアな動作メカニズムを一つずつ紐解いていきます。どうぞお楽しみに。

この記事はWeChatの公開アカウント「Middleware Interest Circle」から転載したものです。以下のQRコードからフォローできます。この記事を転載する場合は、Middleware Interest Circle の公開アカウントにお問い合わせください。