消費者実装ロジック - Kafka 知識システム (IV)

[[410017]]

前回の記事では、Kafka ブローカーの実装原理、データ ストレージ構造、メッセージの永続性について説明しました。メッセージが保存された後、消費者はどのようにそれを消費するのでしょうか?この記事では、Kafka のコンシューマー側について説明します。

1) プル機構

Kafka のプロダクション側では Push というプッシュ メカニズムを使用し、コンシューマー側では Pull というプル メカニズムを使用します。

2) プルのメリットとデメリット

利点は、消費者がメッセージの読み取り速度と量を自分で制御できることです。

デメリットとしては、サーバー上にデータがあるかどうかが分からないため、プルし続けたり、一定間隔でプルしたりする必要があることと、複数回プルして待たなければならない場合があることです。

3) メッセージ配信セマンティクス:

Kafka はデフォルトで少なくとも 1 回の配信を保証し、ユーザーが最大 1 回のセマンティクスを実装できるようにします。正確に 1 回だけ実行する実装は、ターゲット ストレージ システムによって異なります。

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

RangeAssignor: パーティション範囲で割り当てます。現在はデフォルトの戦略です。

RoundRobinAssignor: ラウンドロビン モードでの割り当て。

StickyAssignor: Kafka 0.11 で導入され、負荷などのより多くの指標に基づいて割り当てを可能な限り均等に分散しようとします。

これらについては以前の記事でも触れました。

消費者団体

コンシューマー グループは、Kafka によって提供されるスケーラブルでフォールト トレラントなコンシューマー メカニズムです。 Kafka は Consumer Group メカニズムのみを使用しますが、従来のメッセージ エンジン システムの 2 つの主要モデル、つまりメッセージ キュー モデルとパブリッシュ/サブスクライブ モデルも実装しています。

理想的には、コンシューマー インスタンスの数は、グループがサブスクライブするパーティションの合計数と等しくする必要があります。

消費者と消費者団体

Kafka コンシューマーはコンシューマー グループの一部です。複数のコンシューマーがトピックを消費するためにコンシューマー グループを形成する場合、各コンシューマーは異なるパーティションからメッセージを受信します。 4 つのパーティションを持つ T1 トピックがあるとします。同時に、消費者グループ G1 があり、このグループには消費者 C1 が 1 つだけあります。次に、コンシューマー C1 は、次の 4 つのパーティションからメッセージを受信します。

Kafka の非常に重要な機能は、メッセージを一度だけ書き込むだけでよく、任意の数のアプリケーションがメッセージを読み取ることができることです。つまり、すべてのアプリケーションがメッセージの全量を読み取ることができます。各アプリケーションがメッセージの全量を読み取るためには、アプリケーションに異なるコンシューマー グループが必要です。上記の例で、新しいコンシューマー グループ G2 を追加し、このコンシューマー グループに 2 つのコンシューマーがある場合、次のようになります。

ここで注目すべき点は次のとおりです。

• トピックは複数のコンシューマー グループによって消費される可能性があります。

ただし、各コンシューマー グループによって消費されるデータは互いに干渉しません。つまり、各コンシューマー グループは完全なデータを消費します。

• パーティションは、同じコンシューマー グループ内の 1 つのコンシューマーによってのみ消費されます。

複数の消費者間で分割することはできません。つまり、コンシューマー グループ内のコンシューマーの数がトピックのパーティションの数より多い場合、余分なコンシューマーは機能しません。

消費者パーティション割り当てのプロセス

それでは、割り当てプロセスを見てみましょう。

1. グループコーディネーターを特定する

コンシューマー グループを作成するたびに、Kafka はコンシューマー グループのコーディネーターとしてブローカーを割り当てます。

2. 消費者を登録し、リーダー消費者を選択する

コーディネーターが決まったら、消費者はコーディネーターに登録し始めます。最初に登録した消費者が消費者グループのリーダーになり、後続の消費者はフォロワーになります。

3. リーダーが選出されると、

コーディネータからパーティションとコンシューマの情報をリアルタイムで取得し、パーティション戦略に従って各コンシューマにパーティションを割り当て、割り当て結果をコーディネータに通知します。

4. フォロワー コンシューマーは、消費のためにコーディネータから独自のパーティション情報を取得します。

すべてのフォロワー コンシューマーは、自分が消費するパーティションのみを認識しており、他のコンシューマーの存在は認識していません。

5. この時点で、消費者は全員、自分の消費ゾーンを把握しています。

パーティション分割プロセスは終了しました。パーティションの再バランス調整が発生すると、リーダーは割り当てプロセスを繰り返します。

具体的なフローチャートについては前回の記事をご参照ください。

変位について

【オフセット】

• メッセージを消費するプロセスでは、各コンシューマーは消費したパーティションの現在の位置を記録するフィールドを持っている必要があります。このフィールドは消費者オフセットであり、消費者の消費の進行状況を示す指標です。
• オフセットは値のように見えますが、コンシューマ グループの場合は KV ペアのセットです。キーはパーティションで、V はパーティションを消費するコンシューマの最新の変位に対応します (TopicPartition->long)。
• ただし、コンシューマーの変位は、最後に消費されたメッセージの変位ではなく、次のメッセージの変位であることを覚えておくことが重要です。
• 変位を送信するのは、主に消費者の消費の進捗状況を表すためです。このように、コンシューマーが失敗して再起動すると、以前に送信された変位値を Kafka から読み取り、対応する変位から消費を続行できるため、消費プロセス全体が再度繰り返されるのを回避できます。

[変位保存]

実際、コンシューマ アプリケーションが変位を送信すると、実際にはコーディネーターが配置されているブローカーに変位が送信されます。同様に、コンシューマー アプリケーションが起動すると、コーディネーターが配置されているブローカーにさまざまな要求が送信され、その後、コーディネーターはコンシューマー グループの登録やメンバー管理レコードなどのメタデータ管理操作を実行する責任を負います。

古いバージョンの Consumer Group は ZooKeeper 内の変位を保存します。新しいバージョンの Consumer Group では、Kafka コミュニティは Consumer Group の変位管理方法を再設計し、変位を Kafka の内部トピック、つまり __consumer_offsets (一般に変位トピックと呼ばれる) に保存する方法を採用しました。 Kafka が移転保存のために放棄された理由については、以前の記事「アップグレードされた Kafka Knowledge System 1 の基本概念、アーキテクチャ、および新バージョン」を参照してください。

【変位テーマのデータ形式】

鍵

• 変位トピックのキーには、グループID、トピック名、パーティション番号の3つの部分が含まれている必要があります。

価値

• 保存される主な情報はオフセット情報ですが、もちろんタイムスタンプなどの情報もあります。消費者が消費を開始するポイントを時間に基づいてリセットできることを覚えていますか?

【移転の提出】

1. 自動送信

コミットする最も簡単な方法は、コンシューマーがオフセットを自動的にコミットできるようにすることです。 enable.auto.commit が true に設定されている場合、コンシューマーは 5 秒ごとに poll() メソッドから受信した最大オフセットを自動的にコミットします。

起こりうる問題: データの重複

デフォルトの 5 秒のコミット間隔を引き続き使用すると仮定すると、最後のコミットから 3 秒後に再バランス調整が行われます。再バランス調整後、コンシューマーは最後のコミットのオフセット位置からメッセージの読み取りを開始します。この時点でオフセットはすでに 3 秒遅れているため、この 3 秒以内に到着したメッセージは繰り返し処理されます。コミット間隔を変更してオフセットをより頻繁にコミットし、重複メッセージが表示される可能性のある時間枠を短縮することはできますが、この状況を完全に回避することはできません。

2. 手動送信

2.1 同期送信

同期に関する問題

• 名前が示すように、これは同期操作です。つまり、メソッドは、変位が正常に送信されるまで待機してから戻ります。送信プロセス中に例外が発生した場合、このメソッドは例外情報をスローします。
• commitSync() の問題は、リモート ブローカーがコミット結果を返すまで、コンシューマー プログラムがブロック状態になることです。この状態は終わらないでしょう。コミットが失敗した後、同期コミットが再試行されることに注意してください。
• どのシステムでも、リソースの制限ではなくプログラムによって発生するブロッキングがシステムのボトルネックとなり、アプリケーション全体の TPS とスループットに影響を与える可能性があります。

2.2 非同期送信

手動コミットの欠点の 1 つは、ブローカーがコミット要求に応答するまでアプリケーションがブロックされ、アプリケーションのスループットが制限されることです。コミットの頻度を減らすことでスループットを向上させることができますが、再バランスが発生すると重複メッセージの数が増加します。

このとき、非同期送信を使用して、ブローカーの応答を待たずに送信要求を送信するだけで済みます。再試行しない理由は、サーバーから応答を受信するまでに、より大きなオフセットが正常にコミットされている可能性があるためです。

オフセット 2000 を送信するリクエストを送信するとします。このとき、通信に一時的な問題が発生し、サーバーはリクエストを受信できず、当然応答しません。同時に、別のメッセージのバッチを処理し、オフセット 3000 を正常にコミットしました。commitAsync() がオフセット 2000 でコミットを再試行すると、オフセット 3000 以降で成功する可能性があります。この時点で再バランス調整が発生すると、重複したメッセージが表示されます。

非同期の問題

• commitAsync の問題は、何か問題が発生したときに自動的に再試行されないことです。非同期操作であるため、送信失敗後に自動的に再試行される場合、再試行時に送信された変位値がすでに「期限切れ」になっているか、最新の値ではない可能性があります。したがって、非同期送信を再試行しても意味がないため、プログラムが停止する前に最後の送信が成功している限り、commitAsync は再試行されません。
• 解決策としては、成功か失敗かに関係なくオフセット情報を記録します。最後の送信が成功した場合は無視されます。最後の送信が成功しなかった場合は、次回再起動するときにオフセットを手動で指定できます。

非同期送信と同期送信を組み合わせる

commitSync() と commitAsync() の両方が使用されます。定期的な手動送信の場合、プログラムのブロックを回避するために commitAsync() を呼び出し、コンシューマーが閉じられる前に、commitSync() を呼び出して同期ブロッキング変位送信を実行し、コンシューマーが閉じられる前に正しい変位データが保存されるようにします。

リバランスについて

パーティションの所有権をあるコンシューマーから別のコンシューマーに転送することをリバランスと呼びます。再バランス調整は非常に重要であり、コンシューマー グループに高い可用性とスケーラビリティをもたらし、コンシューマーを安心して追加または削除できます。リバランスをトリガーする 3 つのアクションは次のとおりです。

1. コンシューマーがグループに参加し、元々他のコンシューマーによって読み取られたパーティションを読み取ると、再バランス調整がトリガーされます。
2. コンシューマーがグループを離れる (シャットダウンまたはクラッシュする) と、そのコンシューマーが最初に読み取ったパーティションがグループ内の他のコンシューマーによって読み取られ、再バランスがトリガーされます。
3. 新しいパーティションの追加など、トピックが変更されると、パーティションの再配分が発生し、再バランス調整がトリガーされます。

パーティションの再バランス中はトピックが利用できないため、再バランスが非常に遅くなります。

ここで、不要なパーティションの再バランス調整は、実稼働環境での構成が誤っていることが原因であることを付け加えておきます。

通常のクラスターの変更は考慮されなくなりました。

1. ハートビートを時間内に送信できなかったためにタイムアウトが発生し、コンシューマーがコンシューマー グループから追い出されるのを防ぎます。

ここで、session.timeout.ms タイムアウトと heartbeat.interval.ms ハートビート間隔を設定できます。通常、タイムアウトはハートビート間隔の 3 倍に設定できます。

2. 消費者は時間を費やしすぎている。

コンシューマーが指定された時間内にポーリングからのすべてのメッセージを消費できない場合、コンシューマーに問題があるとみなされ、コンシューマーは独自にグループを離脱します。したがって、max.poll.interval.ms を処理時間よりわずかに長く設定することができます。

3. 2番目の点から、クラスタが頻繁に分割されバランスが取れている場合、

次に、コンシューマーがタスクを実行するのにかかる時間、特に GC にかかる時間を観察する必要があるかもしれません。

多くの場合、オンラインの問題は不合理な構成によって発生します。