Kafka Consumer メッセージの消費と再バランスのメカニズム

Kafka コンシューマー

Kafka にはコンシューマ グループの概念があります。各コンシューマーは、割り当てられているパーティションからのメッセージのみを消費できます。各パーティションは、コンシューマー グループ内の 1 つのコンシューマーのみが使用できます。したがって、同じコンシューマー グループ内のコンシューマーの数がパーティションの数を超えると、一部のコンシューマーには消費するパーティションが割り当てられません。消費者グループと消費者の関係は次の図に示されています。

消費者団体

Kafka コンシューマー クライアントがメッセージを消費するには、通常、次の手順が含まれます。

• クライアントを構成してコンシューマーを作成する
• トピックを購読する
• メッセージを引き出して消費する
• 消費変位を提出する
• コンシューマーインスタンスを閉じる

プロセス

Kafka の Consumer クライアントはスレッドセーフではないため、スレッドの安全性を確保し、消費パフォーマンスを向上させるために、Consumer 側で Reactor に似たスレッド モデルを使用してデータを消費できます。

消費モデル

Kafka コンシューマーパラメータ

• bootstrap.servers: ホスト:ポートの形式でブローカー アドレスに接続します。
• group.id: コンシューマーが属するコンシューマー グループ。
• key.deserializer: プロデューサーの key.serializer (キーのデシリアライズ方法) に対応します。
• value.deserializer: プロデューサーの value.serializer (値のデシリアライズ方法) に対応します。
• session.timeout.ms: コーディネーターが障害を検出するまでにかかる時間。デフォルト値は 10 秒です。このパラメータは、ハートビートの有効期限と同様に、コンシューマー グループが (グループ メンバー comsummer) クラッシュをアクティブに検出する時間間隔です。
• auto.offset.reset: このプロパティは、オフセットのないパーティションまたは無効なオフセット (コンシューマーが長期間非アクティブであり、現在のオフセットが古くて削除されている) を持つパーティションを読み取るときにコンシューマーが実行する操作を指定します。デフォルト値は latest で、これは最新のレコード (コンシューマーの起動後に生成されたレコード) からデータを読み取ることを意味します。もう 1 つの値は earliest です。これは、オフセットが無効な場合に、コンシューマーが開始位置からデータを読み取ることを意味します。
• enable.auto.commit: 変位を自動的にコミットするかどうか。 false の場合は、プログラム内で手動で変位をコミットする必要があります。正確に1回だけのセマンティクスの場合、手動で変位をコミットするのが最善です。
• fetch.max.bytes: 一度に取得するデータの最大バイト数
• max.poll.records: 1 回のポーリング呼び出しで返されるメッセージの最大数。処理ロジックが非常に軽量な場合は、この値を適切に増やすことができます。ただし、max.poll.records 個のデータは session.timeout.ms 内に処理する必要があります。デフォルト値は500です
• request.timeout.ms: リクエスト応答の最大待機時間。タイムアウト期間内に応答が受信されない場合、Kafka はメッセージを再送信するか、再試行回数を超えた場合は直接失敗と見なします。

Kafka の再バランス

リバランスは、本質的には、コンシューマー グループ内のすべてのコンシューマーがサブスクライブされたトピックの各パーティションを割り当てるために合意に達する方法を指定するプロトコルです。たとえば、グループの下に 20 人のコンシューマーがいて、100 個のパーティションを持つトピックをサブスクライブしているとします。通常の状況では、Kafka は各コンシューマーに平均 5 つのパーティションを割り当てます。この割り当てプロセスはリバランスと呼ばれます。

いつリバランスするのでしょうか?

これも、よく言われる質問です。リバランスのトリガー条件は 3 つあります。

• グループ メンバーシップが変更されます (新しいコンシューマーがグループに参加、既存のコンシューマーがグループを離れる、または既存のコンシューマーがクラッシュします。この 2 つの違いについては後で説明します)
• 購読トピックの数が変更されました
• サブスクリプショントピックのパーティション数が変更されました

グループ内でパーティションを割り当てるにはどうすればいいですか?

Kafka はデフォルトで、範囲とラウンドロビンの 2 つの割り当て戦略を提供します。もちろん、Kafka はプラグ可能な割り当て戦略を使用しており、独自のアロケータを作成してさまざまな割り当て戦略を実装できます。

Kafka の面接でよく聞かれる質問

• Kafka ではどのようなコマンドライン ツールが利用できますか?どれを使ったことがありますか？ /bin ディレクトリ、Kafka クラスターの管理、トピックの管理、Kafka の生成と使用。
• Kafka Producer の実行プロセスとは何ですか?インターセプター、シリアライザー、パーティショナー、アキュムレーター。
• Kafka Producer の一般的な構成は何ですか?ブローカー構成、ack 構成、ネットワークおよび送信パラメータ、圧縮パラメータ、ack パラメータ。
• Kafka メッセージを整然とさせるにはどうすればよいでしょうか? Kafka はトピック レベルでは順序付けされておらず、パーティション レベルでのみ順序付けされます。そのため、処理順序を保証するために、パーティショナーをカスタマイズし、順番に処理する必要があるデータを同じパーティションに送信することができます。
• プロデューサーはデータが失われないようにどのように保証しますか? ack メカニズム、再試行メカニズム。
• プロデューサーのパフォーマンスを向上させるにはどうすればいいですか?バッチ、非同期、圧縮。
• 同じグループ内のコンシューマーの数がパーツの数より多い場合、Kafka はどのように処理しますか?冗長パーツは役に立たない状態となり、データを消費しません。
• Kafka Consumer はスレッドセーフですか?安全でない、シングルスレッド消費、マルチスレッド処理。
• Kafka Consumer を使用してメッセージを消費する場合のスレッド モデルについて教えてください。なぜこのように設計されているのでしょうか?引っ張りとハンドリングが分離されています。
• Kafka Consumer の一般的な構成は何ですか?ブローカー、ネットワーク、プル パラメータ、ハートビート パラメータ。
• コンシューマーはいつクラスターから追い出されますか?クラッシュ、ネットワーク異常、処理時間が長すぎるため、送信変位タイムアウトが発生します。
• Consumer が参加または離脱すると、Kafka はどのように反応しますか?リバランスを実行します。
• リバランスとは何ですか? また、いつ行われますか?話題が変われば、消費者も変わります。