私はショックを受けました。これがカフカの「素顔」だったとは！

Kafka は、高いパフォーマンス、永続性、マルチコピーバックアップ、水平スケーラビリティを備えた分散メッセージキューです。プロデューサーはキューにメッセージを書き込み、コンシューマーはキューからメッセージを取得してビジネスロジックを実行します。一般的に、アーキテクチャ設計では、デカップリング、ピークシェービング、非同期処理の役割を果たします。

Kafka はトピックの概念を外部的に使用します。プロデューサーはトピックにメッセージを書き込み、コンシューマーはトピックからメッセージを読み取ります。

水平方向の拡張を実現するために、トピックは実際には複数のパーティションで構成されます。ボトルネックが発生した場合は、パーティションの数を増やすことで水平拡張を行うことができます。メッセージの順序は単一のパーティション内で保証されます。

新しいメッセージが書き込まれるたびに、Kafka はそれを対応するファイルに追加するので、パフォーマンスが非常に高くなります。

Kafka の全体的なデータフローは次のとおりです。

一般的な使用方法は、プロデューサーがブローカー内の指定されたトピックにメッセージを書き込み、コンシューマーがブローカーから指定されたトピックからメッセージをプルして、ビジネス処理を実行することです。

図には 2 つのトピックがあります。 Topic0 には 2 つのパーティションがあり、Topic1 には 1 つのパーティションがあり、3 つのコピーがバックアップされます。

Consumer Gourp1 内の Consumer2 は処理用のパーティションに割り当てられていないことがわかります。これは可能です。以下で説明します。

ZK は、ブローカー、トピック、パーティションに関するメタデータを保存するために使用されます。 ZK は監視とルーティングにも使用されます。

生産

基本的なプロセスは次のとおりです。

レコードを作成します。レコードでは、対応するトピックと値を指定する必要があります。キーとパーティションはオプションです。

まずシリアル化し、次にトピックとパーティションに従って対応する送信キューに入れます。 Kafka Produce はバッチ要求であり、バッチを蓄積してまとめて送信します。 send() を呼び出した後、すぐにネットワークパケットが送信されるわけではありません。

パーティションがいっぱいでない場合、状況は次のようになります。

キーが入力されました。キーに従ってハッシュし、同じキーが 1 つのパーティションに割り当てられます。 (パーティション数が拡張された場合は保証されません。)
キーが入力されていません。ラウンドロビン方式でパーティションを選択します。

同じパーティションに送信されるこれらの要求は、構成に従ってウェーブに蓄積され、別のスレッドによって一度に送信されます。

API

オフセットやルーティングなど、多くのことを実行する高レベル API があり、非常に使いやすいです。

Simple API もあり、Offset などを自分で記録する必要があります。 (注: メッセージを消費するときは、まず消費場所 (ルート) を知っておく必要があります。消費後は、消費順序 (オフセット) を記録する必要があります)

パーティション

複数のコピーがある場合は、異なるブローカーに配布するように努めます。

Kafka はパーティションのリーダーを選択します。その後、パーティションに対するすべてのリクエストはリーダーによって実際に操作され、他のフォロワーに同期されます。

ブローカーが機能を停止すると、そのブローカー上のすべてのリーダーパーティションがリーダーを再選出します。（これは、コピー数を維持するために自動的にレプリケーションが実行される分散ファイルストレージシステムとは異なります）

ここで関係する詳細が 2 つあります。

パーティションの割り当て方法
リーダーの選び方

パーティションの割り当てとリーダーの選出に関しては、必ず実行者が存在しなければなりません。 Kafka では、このエグゼキューターはコントローラーと呼ばれます。

Kafka は、パーティションの割り当てとリーダーの選出のためにブローカー内のコントローラーを選択するために ZK を使用します。

パーティションの割り当て:

すべてのブローカー (合計で n 個のブローカーがあると仮定) と割り当てるパーティションを並べ替えます。
i 番目のパーティションを (i mod n) 番目のブローカー (これがリーダー) に割り当てます。
i番目のパーティションのj番目のレプリカを((i + j)モードn)番目のブローカーに割り当てます。

リーダー災害復旧

コントローラーは、ブローカーがダウンしたかどうかを認識できるように、ZK の /brokers/ids ノードにウォッチを登録します。

ブローカーがダウンすると、コントローラーは影響を受けるパーティションの新しいリーダーを選択します。

コントローラーは、ZK の /brokers/topics/[topic]/partitions/[partition]/state から対応するパーティションの ISR (同期レプリカ) リストを読み取り、リーダーとなるものを 1 つ選択します。

リーダーが選出された後、ZK が更新され、影響を受けるブローカーに変更を通知するために LeaderAndISRRequest が送信されます。

ここで ZK 通知を使用せず、代わりに RPC リクエストをブローカーに直接送信するのはなぜでしょうか?私の理解では、これにより ZK のパフォーマンスに問題が発生する可能性があります。

ISR リストが空の場合、構成に基づいてレプリカがリーダーとしてランダムに選択されるか、パーティションが単に無効になります。

ISR リストにマシンがあっても、そのマシンもダウンしている場合は、ISR マシンが復旧するまで待つことができます。

複数のレプリカの同期

ここでの戦略は、サーバー側の処理として、フォロワーがリーダーからデータを一括して取得して同期するというものです。ただし、具体的な信頼性は製造元によって決まります。

プロデューサーがメッセージを生成すると、データの信頼性は request.required.acks パラメータを通じて設定されます。

Acks=-1 の場合、ISR が min.insync.replicas で指定された数より小さいと、使用不可が返されます。

ISR リスト内のマシンが変更されます。 replica.lag.time.max.ms の設定に従って、一定期間同期が行われないと ISR リストから削除されます。

以前は、ノードは遅れているメッセージの数に基づいて ISR から追い出されていましたが、この値を取得するのが難しく、ピーク時にノードが ISR リストに頻繁に出入りすることが容易であったため、バージョン 1.0 以降は削除されました。

ISA からリーダーが選択されると、フォロワーは自身のログ内の前回の最高水準点以降のレコードを削除し、リーダーに移動して新しいデータを取得します。

新しいリーダーが選出された後、フォロワーのデータが新しいリーダーのデータより多くなる可能性があるため、それを傍受する必要があります。

ここでの最高水準点は、パーティションとリーダーの場合、すべての ISR の最後のレコードを意味します。消費者は最高水準点までしか読み取ることができません。

リーダーの観点から見ると、ハイウォーターマークの更新は 1 ラウンド遅れることになります。たとえば、新しいメッセージが書き込まれると、ISR 内のブローカーはそれをフェッチしますが、ISR 内のブローカーは次のフェッチラウンドでリーダーにのみ通知できます。

まさにこの最高水準点の遅延が原因で、場合によっては Kafka でデータ損失やプライマリデータとスタンバイデータ間の不整合が発生する可能性があります。 0.11 以降では、ハイウォーターマークの代わりにリーダーエポックが使用されます。

考えてみましょう: Acks=-1 の場合

すべてのフォロワーがフェッチに来たときに成功を返す必要がありますか、それともフォロワーによる 2 回目のフェッチを待つ必要がありますか?
リーダーはローカルに書き込まれましたが、ISR 内の一部のマシンで障害が発生しました。どうすればよいでしょうか?

消費

トピックのサブスクリプションはコンシューマーグループに基づいており、コンシューマーグループには複数のコンシューマーを含めることができます。同じコンシューマーグループ内の 2 人のコンシューマーが同時にパーティションを消費することはありません。

つまり、パーティションはコンシューマーグループ内の 1 つのコンシューマーによってのみ消費されますが、複数のコンシューマーグループによって同時に消費される可能性があります。

したがって、コンシューマーグループ内のコンシューマーの数がパーティションの数より多い場合、一部のコンシューマーは常にアイドル状態になります。

API

トピックをサブスクライブするときに、正規表現を使用できます。新しいトピックが一致すると、自動的に購読されます。

オフセット保存

コンシューマーグループがパーティションを消費する場合、消費場所を記録するためにオフセットレコードを保存する必要があります。以前は、これは ZK に保存されていました。 ZK の書き込みパフォーマンスが低いため、以前のソリューションでは、コンシューマーが 1 分ごとにレポートしていました。

ここで、ZK のパフォーマンスは消費速度に重大な影響を及ぼし、繰り返し消費が発生する可能性が非常に高くなります。バージョン 0.10 以降、Kafka はこのオフセットのストレージを ZK から分離し、consumeroffsets topic と呼ばれるトピックに保存します。

メッセージに書き込まれるキーは、Groupid、Topic、Partition で構成され、値はオフセットです。トピックに設定されているクリーンアップ戦略は Compact です。常に最適なキーを保持し、残りは削除します。

一般に、各キーのオフセットはメモリにキャッシュされるため、クエリを実行するときにパーティションをトラバースする必要はありません。キャッシュがない場合、パーティションが最初に走査されてキャッシュが作成され、その後クエリが返されます。

Consumer Group のオフセット情報が書き込まれる consumer_offsets のパーティションを決定します。具体的な計算式は以下のとおりです。

 __consumers_offsets パーティション =
           数学。 abs (groupId.hashCode() % groupMetadataTopicPartitionCount)
 //groupMetadataTopicPartitionCount は offsets.topic.num.partitions によって指定され、デフォルトは 50 パーティションです。

考えてみてください。実行中のサービスが offsets.topic.num.partitions を変更すると、Offset のストレージが台無しになるでしょうか?

パーティションの割り当て - 再バランス

生成プロセス中、ブローカーはパーティションを割り当てる必要があり、消費プロセス中、パーティションはコンシューマーにも割り当てられる必要があります。

ブローカーがコントローラーを選択するのと同じように、コンシューマーもブローカーからコーディネーターを選択してパーティションを割り当てます。

上から下への説明は以下の通りです。

コーディネーターの選び方
対話プロセス
リバランスプロセス

① コーディネーターを選択: オフセットが保存されているパーティションを確認します。パーティションリーダーが配置されているブローカーが選択されたコーディネーターです。

ここで、コンシューマーグループのコーディネーターとコンシューマーグループオフセットを格納するパーティションリーダーが同じマシンであることがわかります。

②インタラクションプロセス：コーディネーターを選択したら、割り当てます。全体のプロセスは次のとおりです。

コンシューマーが起動するかコーディネーターがダウンすると、コンシューマーはブローカーをランダムに要求し、ConsumerMetadataRequest リクエストを送信します。

ブローカーは、上記の方法に従って、このコンシューマーに対応するコーディネーターのアドレスを選択します。

コンシューマーはコーディネーターにハートビート要求を送信します。 IllegalGeneration が返された場合、コンシューマーの情報が古く、再結合して Rebalance を実行する必要があることを意味します。

結果が成功した場合、コンシューマーは最後に割り当てられたパーティションから実行を続行します。

③リバランスプロセス：

コンシューマーはコーディネーターに JoinGroupRequest リクエストを送信します。
このとき、他のコンシューマーがハートビート要求を送信すると、コーディネーターは再バランスが必要であることを伝えます。
他のコンシューマーは JoinGroupRequest リクエストを送信します。
記録されたすべてのコンシューマーが JoinGroupRequest リクエストを送信すると、コーディネーターはこれらのコンシューマーからリーダーをランダムに選択します。
次に、JoinGroupResponse を返します。これにより、コンシューマーに自分がフォロワーであるかリーダーであるかが通知されます。リーダーの場合、フォロワーの情報も取得され、この情報に基づいてパーティションを割り当てることができます。
コンシューマーはコーディネーターに SyncGroupRequest を送信します。リーダーの SyncGroupRequest には割り当て情報が含まれます。
コーディネーターは、リーダーを含むコンシューマーに割り当て状況を通知するために返信します。

パーティションまたはコンシューマーの数が変更された場合は、再バランス調整を実行する必要があります。

リバランスが発生する可能性がある状況は次のとおりです。

パーティションの追加
消費者を増やす
消費者は積極的に閉じる
消費者のダウンタイム
コーディネーター自体がダウンしている

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

Kafka は 3 つのメッセージ配信セマンティクスをサポートしています。

最大 1 回: 最大 1 回、メッセージは失われる可能性がありますが、重複することはありません。
少なくとも 1 回: 少なくとも 1 回、メッセージは失われませんが、繰り返される可能性があります。
正確に 1 回: 1 回のみ、メッセージは失われたり重複したりせず、1 回だけ消費されます (0.11 で実装され、ダウンストリームと Kafka に限定されています)

ビジネスでは、「少なくとも 1 回」モデルがよく使用されます。再入可能性が必要な場合は、通常、企業が自らそれを実装します。

少なくとも1回

最初にデータを取得し、次にビジネスを処理し、ビジネス処理が成功した後にオフセットをコミットします。

プロデューサーが異常なメッセージを生成します。メッセージが正常に書き込まれたかどうかは不明です。メッセージをやり直すと、重複したメッセージが書き込まれる可能性があります。
コンシューマーはメッセージを処理します。ビジネス処理が成功した後、オフセットの更新は失敗します。コンシューマーが再起動されると、消費が繰り返されます。

最大1回

まずデータを取得し、次にオフセットをコミットし、最後にビジネス処理を実行します。

プロデューサーが異常なメッセージを生成します。何が起こっても、次のメッセージが生成され、メッセージは失われます。
コンシューマーがメッセージを処理するときは、まずオフセットを更新し、次にビジネス処理を実行します。ビジネス処理が失敗し、コンシューマーが再起動されると、メッセージは失われます。

正確に1回

まずメッセージが失われないようにし、次にメッセージが繰り返されないようにするというのが目的です。それでは、「少なくとも一度は」の理由に注目してみましょう。

最初に思いついたのは:

プロデューサーのやり直しにより、メッセージが重複して書き込まれることになります。プロダクションでは冪等性が保証されます。
消費者による重複消費: 重複消費を排除するか、重複消費が問題にならないようにビジネスインターフェイスのべき等性を確保します。

Kafka はビジネスインターフェースがべき等であるかどうかを保証できないため、ここで Kafka が提供する Exactly once は制限されており、コンシューマーのダウンストリームも Kafka である必要があります。

したがって、以下の説明では、特に指定がない限り、コンシューマーの下流システムは Kafka です (注: 一部のシステムを適応させて Exactly once を実現する Kafka Conector が使用されます)。プロデューサーの冪等性は実装が簡単で、問題はありません。

重複消費を解決するには 2 つの方法があります。

ダウンストリームシステムは冪等性を保証し、繰り返し使用しても複数のレコードが生成されることはありません。
コミットオフセットとビジネス処理をトランザクションにバインドします。

本来であれば、ポイント 1 を 1 回だけ達成すれば問題ありません。ただし、使用シナリオによっては、データソースが複数のトピックであり、処理されて複数のトピックに出力されることがあります。この場合、出力が完全に成功するか、完全に失敗するかのいずれかになることを期待します。これにはトランザクション性が必要です。

トランザクションを実行する必要があるため、ルートからの重複消費の問題を解決し、コミットオフセットと他のトピックへの出力を 1 つのトランザクションにバインドできます。

生成冪等性

アイデアとしては、各プロデューサーにプロデューサーの一意の識別子として Pid を割り当てることです。

プロデューサーはそれぞれに対して単調に増加する Seq を維持します。同様に、ブローカーはそれぞれに対して *** Seq も記録します。

ブローカーは、req_seq == broker_seq+1 の場合にのみメッセージを受け入れます。理由は次の通りです。

メッセージの Seq が Broker Seq より大きい場合、一部のデータがまだ書き込まれていない、つまり順序が正しくないことを意味します。
メッセージの Seq がブローカーの Seq より小さくない場合、メッセージは保存されています。

トランザクション/アトミックブロードキャスト

シナリオは次のとおりです。

まず、複数のソーストピックからデータを取得します。
ビジネス処理を実行し、複数のダウンストリーム宛先トピックに書き込みます。
複数のソーストピックのオフセットを更新します。

ポイント 2 と 3 は、完全に成功するか完全に失敗するトランザクションと見なされます。これは、オフセットが実際には特別なトピックに保存されるためです。これら 2 つのポイントは、複数のトピックを書き込むトランザクション処理に統合されます。

基本的な考え方は次のとおりです。

Tid (トランザクション ID) の紹介。 Pidとは異なり、このIDはアプリケーションによって提供され、トランザクションを識別するために使用されます。プロデューサーが誰であるかは関係ありません。

つまり、どのプロデューサーもこの Tid を使用してトランザクションを実行できるため、途中で終了したトランザクションは別のプロデューサーによって回復できます。

同時に、トランザクションの状態を記録するために、Offset の処理と同様に、トランザクションログを記録する Transaction Coordinator が導入されています。

クラスターには複数のトランザクションコーディネーターが存在し、各 Tid は一意のトランザクションコーディネーターに対応します。

注: トランザクションログの削除戦略は Compact です。完了したトランザクションは Null としてマークされ、Compact 後に保持されません。

トランザクションを実行するときは、まずトランザクションをオープンとしてマークし、データを書き込み、すべてが成功した場合は、トランザクションログに Prepare Commit ステータスとして記録し、それ以外の場合は Prepare Abort ステータスとして書き込みます。

次に、関連する各パーティションにマーカー (コミットまたは中止) メッセージを書き込み、このトランザクションのメッセージを読み取り可能または破棄済みとしてマークします。成功すると、コミット/中止ステータスがトランザクションログに記録され、トランザクションが終了します。

データフロー:

まず、Tid を使用して、任意のブローカー (負荷が最も小さいブローカーがコードに記述されています) に、対応するトランザクションコーディネーターを検索するように要求します。
トランザクションコーディネータに、対応する Pid と Pid に対応するエポックを取得するように要求します。このエポックは、ゾンビプロセスが復活してメッセージの混乱が発生するのを防ぐために使用されます。

メッセージのエポックが現在維持されているエポックよりも小さい場合、そのメッセージは拒否されます。 TidとPidは1対1に対応しているので、同じTidに対しては同じPidが返されます。

クライアントはまず、トランザクションコーディネータによって記録されたトランザクションステータスを要求します。初期ステータスは「開始」です。トランザクション内で最初に到着した場合、トランザクションは同時に計測されます。

クライアントは関連するパーティションにデータを出力します。次に、クライアントはトランザクションコーディネーターにオフセットのトランザクションステータスを記録するように要求します。クライアントは、対応するオフセットパーティションにオフセットコミットを送信します。

クライアントはコミット要求を送信し、トランザクションコーディネーターはコミット/中止の準備を記録し、関連するパーティションにマーカーを送信します。

すべてが成功すると、コミット/中止ステータスが記録されます。このレコードは、Prepare が失われない限り最終的な正確性が保証されるため、他のレプリカの ACK を待つ必要はありません。

ここでの準備ステータスは、主にトランザクションの回復に使用されます。たとえば、制御メッセージが該当するパーティションに送信され、完了する前にマシンがクラッシュした場合、スタンバイマシンが起動した後、プロデューサーが Pid を取得する要求を送信すると、未完了のトランザクションが完了します。

コミットマーカーがパーティションに書き込まれると、関連するメッセージを読み取ることができます。したがって、Kafka トランザクションの Prepare Commit から Commit までの期間では、メッセージは同時に表示されるのではなく、徐々に表示されます。

消費者問題

上記はすべて制作の観点から見たものです。消費の観点からもいくつかの問題を考慮する必要があります。

消費中、パーティション内にはコミットされていない状態、つまりビジネス側では表示されないメッセージがいくつかあります。これらのメッセージは、ビジネス側で表示されないようにフィルタリングする必要があります。 Kafka は、主にパフォーマンスを考慮して、ブローカーでフィルタリングするのではなく、コンシューマープロセスでこれを行うことを選択します。

Kafka の高パフォーマンスの重要なポイントはゼロコピーです。ブローカーでフィルタリングが必要な場合は、メッセージの内容をメモリに読み込む必要があり、ゼロコピー機能が失われます。

ファイル構成

Kafka のデータは、実際にはファイルシステムにファイルの形式で保存されます。トピックの下にパーティションがあり、パーティションの下にセグメントがあります。セグメントは実際のファイルですが、トピックとパーティションは抽象的な概念です。

実際のログファイル (つまり、セグメント) と対応するインデックスファイルは、ディレクトリ /partitionid}/ に保存されます。

各セグメントファイルのサイズは同じで、ファイル名はセグメント内の最小のオフセットに基づいて命名されます。ファイル拡張子は.logです。セグメントに対応するインデックスのファイル名は同じで、拡張子は .index です。

インデックスファイルは 2 つあります。

1 つはオフセットインデックスで、オフセットによってメッセージを検索するために使用されます。
1 つは時間による検索に使用される時間インデックスです。実際、それらを最適化して組み合わせることもできます。以下ではオフセットインデックスについてのみ説明します。

全体的な構成は次のとおりです。

インデックスファイルのサイズを縮小し、スペースの使用量を減らし、メモリへの直接読み込みを容易にするために、ここでのインデックスではスパースマトリックスを使用します。各メッセージの特定の場所を記録する代わりに、一定のバイト数ごとにインデックスが作成されます。

インデックスは 2 つの部分で構成されます。

BaseOffset: これは、このインデックスが対応するセグメントファイル内のメッセージの数を意味します。これにより、数値圧縮アルゴリズムを使用してスペースを節約することが容易になります。たとえば、Kafka は Varint を使用します。
位置: セグメント内の絶対位置。

オフセットに対応するレコードを検索する場合、まずバイナリ検索を使用して、対応するオフセットがどのセグメントにあるかを調べます。次に、インデックスを使用してセグメント内のオフセットのおおよその位置を特定し、メッセージを走査します。

一般的な設定項目

ブローカー構成

トピックの設定

ログのクリーニングに関しては、デフォルトでは、現在書き込まれているログはクリーンアップされません。

0.10 より前のバージョンでは、時間はログファイルの Mtime に基づいていますが、この値は不正確です。ファイルが変更されると、Mtime が変更される場合があります。したがって、バージョン 0.10 以降では、ファイルが最後に送信された時刻を使用して時間が決定されます。

サイズ別にクリーンアップします。ここで注目すべき点は、Kafka がスケジュールされたタスク内の現在のログボリュームの合計サイズを比較して、少なくとも 1 つのログセグメントのしきい値を超えているかどうかを確認しようとしていることです。 1 つのログセグメントを超えているが、超えていない場合は削除されません。

著者: 鄭潔文

紹介：Tencent Cloud Storage、シニアバックエンドエンジニア。2014年に卒業後、Tencentに入社し、付加価値ビジネス開発とTencent Cloud Storage開発に従事。当社は、ビジネスおよび技術プラットフォームのバックエンドアーキテクチャの設計について、徹底的な調査と実践を行っています。彼は、アーキテクチャの大規模な同時実行性、高可用性、スケーラビリティに関して豊富な経験を持ち、現在は分散およびストレージ分野に重点を置いています。

<<: 分散トレーニング入門: PyTorch を使用してマルチ GPU 分散トレーニングを実装する方法

>>: UiPathはシリーズDの資金調達で5億6800万ドルを獲得し、評価額は70億ドルを超える