Kafka の 3 階層アーキテクチャ設計の分析

[[432602]]

この記事はWeChatの公開アカウント「Hua Zai Chats about Technology」から転載したもので、著者はWang Jianghuaです。この記事を転載する場合は、Huazai Chat Technology パブリックアカウントにお問い合わせください。

1 Kafkaの3つの高レベルアーキテクチャの概要

最近は仕事や色々なことが忙しくて、この記事を書くのが大変でした。数週間にわたる考えと整理を経て、ついに完成しました。前回の記事では、Kafka の基礎、ワークフロー、ストレージの仕組み、レプリケーションなどの知識について説明しました。この記事では、Kafka の高可用性、高パフォーマンス、高同時実行性アーキテクチャ設計の秘密を明らかにします。

Kafka は、高スループット、低レイテンシ、高同時実行性、高スケーラビリティで常に知られており、ますます多くのシナリオで使用されています。現時点では、安定性に対する要件はより高くなっています。それでは、詳細を一つずつご紹介させていただきます。

2 Kafka の高可用性設計

リーダー選出メカニズム

Kafka の選挙は、コントローラー選挙、リーダー選挙、コンシューマー選挙の 3 つのカテゴリに大別できます。リーダー選出について説明する前に、まず Kafka コントローラー、つまりブローカーについて説明しましょう。一般的なブローカー機能に加えて、パーティションリーダーノードを選出する機能も備えています。 Kafka システムが起動すると、ブローカーの 1 つがコントローラーとして選出され、トピックパーティションとレプリカのステータスの管理と再配布タスクの実行を担当します。

コントローラーの起動シーケンスは次のとおりです。

1) 最初に起動されたノードは、Zookeeper システムに一時的なノード/コントローラーを作成し、ノードの登録情報を書き込んでノードをコントローラーにします。

2) 他のノードが次々に起動されると、それらも Zookeeper システム内に /controller ノードを作成しようとしますが、/controller ノードがすでに存在するため、「/controller ノード例外の作成に失敗しました」というメッセージがスローされます。作成に失敗したノードは、返された結果に基づいて Kafka クラスター内にコントローラーが正常に作成されたと判断し、/controller ノードの作成を中止します。これにより、Kafka クラスターコントローラーの一意性が確保されます。

3) 他のノードもコントローラーに対応するリスナーを登録し、各リスナーは自身のプロキシノードのステータスの変化を監視する責任を負います。ノードのステータスの変化が検出されると、対応する監視機能がトリガーされ、処理が行われます。

コントローラーの知識について説明した後、リーダーノードを選出するプロセスについて説明しましょう。コントローラーを選出する際の基本的な考え方は、各ノードが Zookeeper システム内の /controller 一時ノードの作成を獲得するために公平に競争することです。最初に正常に作成されたノードはコントローラーになり、トピックパーティションのリーダーノードを選出する機能を持ちます。選挙プロセスは以下の図に示されています。

コピーメカニズム

簡単に言えば、レプリカメカニズムとは、分散クラスターに同じデータのバックアップを保存するバックアップメカニズムです。レプリケーションメカニズムの利点は、データの冗長性を提供することです。レプリケーションメカニズムは、システムの高可用性と高耐久性を確保するための Kafka の重要な基礎です。

高可用性を確保するために、Kafka パーティションが複製されます。レプリカの 1 つが失われた場合、パーティションデータは他のレプリカから取得できます (対応するレプリカのデータは完全である必要があります)。これが Kafka データの一貫性の基礎となります。以下では、Kafka のレプリケーションメカニズムについて詳しく説明します。

Kafka は Zookeeper を使用してクラスターブローカーの情報を維持します。各ブローカーには一意の識別子 broker.id があり、クラスター内で自身を識別するために使用されます。ブローカーは Zookeeper を通じてコントローラーブローカーと呼ばれるノードを選択します。他のブローカーの機能に加えて、トピックパーティションとそのレプリカのステータスの管理も担当します。

Kafka では、トピックは複数のパーティションに分割されます。パーティションは、Kafka の最も基本的なストレージユニットです。トピックを作成するときに、replication-factor パラメータを使用してパーティションのコピー数を指定できます。パーティションレプリカには常にリーダーコピーが存在します。すべてのメッセージはリーダーのコピーに直接送信されます。データの一貫性を確保するために、他のレプリカはリーダー内のデータをコピーする必要があります。リーダーのレプリカが利用できない場合は、フォロワーの 1 人が選出され、新しいリーダーになります。

ISRメカニズム

ISRを理解する

上の図に示すように、各パーティションには ISR (同期レプリカ) リストがあり、同期された利用可能なレプリカをすべて維持するために使用されます。リーダーレプリカは同期レプリカである必要があります。つまり、ISR はフォロワーレプリカのセットだけではなく、リーダーレプリカも含まれます。場合によっては、ISR にはリーダーという 1 つのレプリカのみが存在することがあります。フォロワーレプリカの場合、同期レプリカと見なされるには次の条件を満たす必要があります。

1) ハートビートを Zookeeper に定期的に送信する必要があります。

2) 指定された時間内に「低遅延」でリーダーレプリカからメッセージを取得します。

レプリカが上記の条件を満たさない場合、ISR リストから削除され、条件が満たされるまで再度追加されることはありません。そのため、フォロワーがリーダーとリアルタイムに同期できないリスクが生じる可能性があります。

Kafka がフォロワーがリーダーと同期されているかどうかを判断する条件は、ブローカー側のパラメータ replica.lag.time.max.ms の値です。このパラメータは、フォロワーコピーがリーダーコピーより遅れることができる最大時間間隔を意味します。現在のデフォルト値は 10 秒です。つまり、フォロワーコピーがリーダーコピーより 10 秒以上遅れ続けることがない限り、フォロワーコピーがリーダーコピーよりも少ないメッセージを保持している場合でも、Kafka は 2 つが同期されていると見なします。

Kafka での ISR の管理は、最終的に Zookeeper ノードにフィードバックされます。具体的な場所は、/brokers/topics/[topic]/partitions/[partition]/state です。現在、Zookeeper ノードが保守されている場所は 2 か所あります。

1) コントローラーのメンテナンス: Kafka クラスター内のブローカーの 1 つがコントローラーとして選出されます。コントローラーは主にパーティション管理とレプリカステータス管理を担当し、パーティションの再配布などの管理タスクも実行します。特定の条件下では、コントローラーの下の LeaderSelector が新しいリーダーを選出し、ISR と新しい leader_epoch および controller_epoch が Zookeeper の関連ノードに書き込まれます。同時に、leaderAndIsrRequest が開始され、すべてのレプリカに通知されます。

2) リーダーのメンテナンス: リーダーには、ISR 内のフォロワーが ISR を離れたかどうかを定期的に検出するための別のスレッドがあります。 ISR が変更されると、新しい ISR 情報が Zookeeper の関連ノードに返されます。

ACKメカニズム

acks パラメータは、Kafka の使用において中核となる重要なパラメータです。それは多くのことを決定します。 ack は、レプリケーションメカニズム、同期メカニズム、および ISR メカニズムと密接に関連しています。これらを理解できない場合は、acks パラメータの意味を完全に理解することはできません。

まず、プロデューサークライアントである KafkaProducer に acks パラメータを設定します。つまり、Kafka にデータを書き込むときに、acks パラメータを設定できます。このパラメータには、実際には 0、1、およびすべての 3 つの共通値を設定できます。

確認 = 0

acks が 0 に設定されている場合、プロデューサーはブローカーからのフィードバックを待機しません。メッセージはすぐにソケットバッファーに追加され、送信されたとみなされます。この場合、サーバーがリクエストを受信したという保証はなく、Retries パラメータは有効になりません (クライアントが失敗情報を取得できないため)。

このとき、各レコードに対して返されるオフセットは常に -1 に設定されます。このモードでは、Kafka は最高のスループットと同時実行性を実現しますが、データが失われやすくなります。これは通常、アプリケーションログを記録し、データ要件が低い一部のビジネスシナリオに適しています。

確認 = 1

acks が 1 に設定されている場合、リーダーノードは最初にレコードをローカルログに書き込み、すべてのフォロワーノードがフィードバックする前に成功を確認します。この場合、レコードを受信した後、フォロワーノードがデータの複製を完了する前にリーダーノードでエラーが発生すると、レコードは失われます。このモードは、MySQL のマスタースレーブ非同期レプリケーションと同じです。マスターとスレーブの間にはデータの違いが生じます。この構成は Kafka のデフォルト構成です。データのセキュリティとパフォーマンスのバランスをとります。

acks = すべて & min.insync.replicas >= 2

acks が all に設定されている場合、リーダーノードは、レコードが送信されたかどうかを確認する前に、すべての同期された LSR コピーからの確認を待機します。少なくとも 1 つの同期レプリカが存在する限り、レコードは失われません。

リーダーがメッセージを受信したが、フォロワーがメッセージを受信せず、リーダーがダウンしている場合、クライアントはメッセージが正常に送信されなかったことを感知し、再度メッセージの送信を試行します。

ブローカーには、プロデューサーデータを正常に書き込むために必要な ISR の最小数を表す構成項目 min.insync.replicas (デフォルト値は 1) があります。 ISR 内のレプリカの数が min.insync.replicas より少ない場合、リーダーはプロデューサーによって生成されたメッセージの書き込みを停止し、プロデューサーに NotEnoughReplicas 例外をスローして、より多くのフォロワーが追いついて ISR に再び入るのをブロックして待機します。したがって、min.insync.replicas-1 個のレプリカが同時に失敗しても許容できます。

この方法はパフォーマンスを犠牲にしますが、データ要件が高いビジネスシナリオに適しています。

3 Kafkaの高性能設計

原子炉多重化モデル

Reactor (多重化) について話すとき、Java の NIO について言及する必要があります。次に、Java の NIO を見てみましょう。

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

1) チャネル

2) バッファ

3) セレクター

チャネルは、データストリームを送信するために使用される Java のストリームと同じです。次の図に示すように、チャネルからバッファにデータを読み取ることも、バッファからチャネルにデータを書き込むこともできます。

セレクターを使用すると、単一のスレッドで複数のチャネルを処理できます。セレクタを使用するには、まずセレクタにチャネルを登録し、その select() メソッドを呼び出す必要があります。このメソッドは、登録されたチャネルでイベントが準備されるまでブロックされます。このメソッドが返されると、スレッドはこれらのイベント（新しい接続の受信、データの受信など）を処理できます。

次の図は、セレクターを使用して 3 つのチャネルを処理する単一のスレッドを示しています。

Kafka SocketServer は Java NIO に基づいて開発されており、Reactor モデル (実際に非常に効率的であることが証明されており、Netty や Mina で広く使用されています) を採用しています。 Kafka Reactor モデルは、次の 3 つのロールで構成されます。

1) アクセプター

2) プロセッサ

3) ハンドラー

Kafka Reactor には、クライアント要求を受け入れる 1 つの Acceptor、データの読み取りと書き込みを行う N 個の Processor スレッド (つまり、各接続ごとに個別に処理する Processor が作成され、各 Processor は独立した Selector を参照します)、およびビジネスロジックを処理する M 個の Handler が含まれます。アクセプタとプロセッサの間、およびプロセッサとハンドラの間には、リクエストをバッファリングするためのキューがあります。

次の図は、Kafka の簡略化された Reactor モデルアーキテクチャを示しています。

生産メッセージプロセス

プロデューサーが Kafka クラスターに送信する詳細なプロセスを次の図に示します。

1) まず、メッセージが届くと、プロデューサーソースコードはメッセージを ProducerRecord オブジェクトにカプセル化します。

2) オブジェクトにカプセル化された後、オブジェクトはシリアル化され（ネットワーク転送を含む）、シリアル化のために Serializer コンポーネントが呼び出され、シリアル化後に送信されます。

3) 送信する前に、どのトピックのどのパーティションにメッセージを送信するかを決定する必要があります。この時点で、Partitioner パーティショナーを介して Kafka Broker クラスターからクラスターメタデータを取得する必要があります。メタデータを取得したら、送信できます。

4) バージョン 0.8 より前では、メッセージが届くと、リクエストにパッケージ化されてブローカーに送信されます。この場合、パフォーマンスは非常に悪くなります。バージョン0.8以降では簡単な改良が加えられ、パフォーマンスが飛躍的に向上しました。つまり、メッセージが到着すると、すぐに送信されるのではなく、まずキャッシュ (RecordAccumulator) キューに書き込まれ、その後バッチ (RecordBatch) にパッケージ化されます。

5) このとき、送信スレッドは複数のバッチを 1 つのリクエストにカプセル化して送信します。これにより、多くのリクエストが削減され、スループットが向上します。現時点では問題があります。メッセージは受信されるとすぐに送信されるのではなく、バッチにカプセル化されます。遅延の問題はありますか?デフォルトの batch.size は 16K です。満杯の場合はすぐに発送いたします。いっぱいでない場合は、指定された時間に送信されます（linger.ms = 500ms）

6) 送信時に、各リクエストリクエストはマルチプレクサ（セレクタ）内の各Kafkaチャネルに対応し、その後ブローカークラスタにデータを送信します。

7) Batch バッチと Request リクエストをカプセル化するプロセスでは、重要な設計概念であるメモリプールソリューションも関係します。これについては、以下のサーバーメモリプールのセクションで詳しく説明します。

ディスクへのシーケンシャル書き込み + OS キャッシュ

まず、ディスク書き込みパフォーマンスを確保するために、Kafka はオペレーティングシステムに基づいてページキャッシュを介したファイル書き込みを実装します。オペレーティングシステム自体には、ページキャッシュと呼ばれるキャッシュレイヤーがあり、これはメモリ内のキャッシュです。これを OS キャッシュと呼ぶこともできます。これは、オペレーティングシステム自体によって管理されるキャッシュを意味します。次に、ディスクファイルを書き込むときに、最初に OS キャッシュに直接書き込むことができます。つまり、メモリに書き込むだけです。次に、オペレーティングシステムは、OS キャッシュ内のデータを実際にディスクにフラッシュするタイミングを決定し、書き込みの効率とパフォーマンスが大幅に向上します。次の図に示すように:

もう 1 つの非常に重要な操作は、Kafka がデータを書き込むときに、ディスクシーケンシャル方式でディスクに書き込むことです。つまり、ファイル内のランダムな位置でデータを変更するのではなく、ファイルの末尾にデータを追加します。通常の機械式ディスクの場合、ランダムに書き込まれるとディスクのアドレス指定の問題が発生し、パフォーマンスが極端に低下します。ただし、ファイルの末尾に順番に追加していくだけであれば、このディスクのシーケンシャル書き込みのパフォーマンスは、基本的にメモリへの書き込みのパフォーマンスと同じです。

ゼロコピー技術

上記は執筆プロセスについての説明です。それでは消費プロセスについてお話ししましょう。 Kafka からのデータの使用には、実際には Kafka のディスクファイルからデータを読み取り、それを下流のコンシューマーに送信することが含まれます。一般的なプロセスは次のとおりです。

1) まず、読み取るデータが OS キャッシュにあるかどうかを確認します。そうでない場合は、ディスクファイルからデータを読み取り、OS キャッシュに格納します。

2) 次に、OS キャッシュからアプリケーションプロセスのキャッシュにデータをコピーし、アプリケーションプロセスのキャッシュからオペレーティングシステムレベルのソケットキャッシュにデータをコピーし、最後にソケットキャッシュからデータを読み取ってネットワークカードに送信し、最後にネットワークカードから下流のコンシューマーに送信します。

上の図からわかるように、プロセス全体で2つの不要なコピー操作があります。

1) オペレーティングシステムの OS キャッシュからアプリケーションプロセスのキャッシュにデータをコピーします。

2) 次に、アプリケーションキャッシュからオペレーティングシステムのソケットキャッシュにコピーされます。

これら 2 つのコピー中に、複数のコンテキストスイッチが発生したため、パフォーマンスが比較的集中します。

この問題を解決するために、Kafka はデータの読み取り時にゼロコピー技術を導入しました。つまり、オペレーティングシステムの OS キャッシュ内のデータは、ネットワークカードに直接送信され、その後下流の消費者に転送されるため、途中でデータをコピーする 2 つの手順が省略され、コピーの CPU オーバーヘッドと、ユーザーモードとカーネルモード間のコンテキスト切り替えの回数が削減され、データ転送のパフォーマンスが最適化されます。ソケットキャッシュでは、記述子が 1 つだけコピーされ、データはソケットキャッシュにコピーされません。次の図に示すように:

一般的なゼロコピーの考え方を実装するには、主に 2 つの方法があります。

1) 直接 I/O: データはカーネルを直接スキップし、ユーザー空間と I/O デバイス間で転送されます。この場合、カーネルは必要な補助作業のみを実行します。

2) コピーオンライト: コピーオンライトでは、データを事前にコピーする必要はなく、変更が必要なときにデータの一部のみがコピーされます。

ここで、Kafka は主に mmap と sendfile を使用してゼロコピーを実現します。これは、Java の MappedByteBuffer と FileChannel.transferIO に相当します。

Java NIO を使用して実装されたゼロコピーは次のとおりです。

transferTo() メソッドは、ファイルチャネルから指定された書き込み可能なバイトチャネルにデータを転送します。内部的には、基盤となるオペレーティングシステムのゼロコピーサポートに依存します。 Linux システムでは、この呼び出しは sendfile() システムコールに渡され、次のように処理されます。

圧縮伝送

デフォルトでは、Kafka プロデューサーでは圧縮は有効になっていません。これにより、プロデューサーからブローカーへの転送が高速化されるだけでなく、レプリケーション中の転送も高速化されます。圧縮により、スループットが向上し、待ち時間が短縮され、ディスク使用率が向上します。

Kafka では、圧縮はプロデューサー側とブローカー側の 2 つの場所で発生する可能性があります。圧縮と解凍を一言でまとめると、プロデューサー側が圧縮し、ブローカー側が維持し、コンシューマー側が解凍します。

Kafka は、lz4、snappy、gzip など複数の圧縮アルゴリズムをサポートしています。 Kafka 2.1.0 以降では、ZStandard アルゴリズムが追加されました。このアルゴリズムは Facebook のオープンソース圧縮アルゴリズムであり、超高圧縮率を実現できます。

プロデューサー、ブローカー、コンシューマーは同じ圧縮アルゴリズムを使用する必要があります。プロデューサーがブローカーにデータを書き込み、コンシューマーがブローカーからデータを読み取る場合、データを解凍する必要はありません。解凍は、コンシューマーが最終的にメッセージを受信したときにのみ必要になります。これにより、ネットワークとディスクのオーバーヘッドを大幅に削減できます。

サーバーメモリプールの設計

先ほど、バッチとリクエストを含むメッセージ生成の詳細なプロセスについて説明しました。このプロセスにおいて、Kafka にはメモリプールソリューションという重要な設計概念もあります。ここではメモリプールの実装プロセスについて詳しく説明します。

1) ここではプロセスを簡略化します。最初にメッセージがカプセル化され、シリアル化され、最後にパーティション番号が計算され、メッセージがキャッシュに保存されます。

2) このキャッシュにはバッチキューもあります。では、このバッチキューを保存するためにどのような戦略が使用されるのでしょうか? 1 つのパーティションは 1 つのキューに対応します。ここには重要なデータ構造、バッチがあります。このデータ構造はキーと値の形式です。キーはメッセージトピックのパーティションであり、値はキューです。対応するパーティションに送信されたバッチはそこに保存されます。

3) 2 つのトピックがあり、それぞれに 2 つのパーティションがある場合、合計で 4 つのパーティション、つまり 4 つのキューがあり、それぞれにバッチがあることになります。メッセージは計算され、分割された後、キューの最新のバッチに書き込まれます。

4) 送信スレッドは、バッチがいっぱいかどうか、または時間が来たかどうかを確認します。条件が満たされると、Sender スレッドはそれを取り出して Request にカプセル化し、送信します。

5) バッチのパッケージ化ではメモリが使用され、送信者が送信した後にメモリはリサイクルされます。 Java でメモリ操作やリサイクルを頻繁に行うと、FullGC の頭痛の種となり、ワーカースレッドのパフォーマンスが低下し、プロデューサー全体のパフォーマンスに影響を及ぼします。 Kafka のソリューションはメモリプールであり、メモリブロックの使用方法はデータベース接続プールと同じです。

6) バッファポールメモリプール全体のサイズは 32M です。メモリプールは 2 つの部分に分かれています。 1 つの部分はメモリキューで、メモリブロック (16K) が含まれています。残りの部分は使用可能なメモリです。メッセージが到着すると、メモリプールからメモリブロックが適用されます。適用後、バッチがカプセル化され、データが書き込まれます。送信スレッドは送信と応答を行い、その後メモリをクリアして繰り返し使用できるようにメモリプールに戻します。これにより、GC の頻度が大幅に削減され、プロデューサーの安定性と効率性が確保され、パフォーマンスが大幅に向上します。

4 Kafkaの高並列設計

高同時実行ネットワーク設計

上記のセクションでは、Kafka プロデューサーとサーバーの高可用性と高パフォーマンスについて説明しました。ここでは、Kafka で最も古典的な、Kafka の超高同時実行ネットワークアーキテクチャ設計を主に分析します。

ここでは、上図に示すように、Kafka のネットワークアーキテクチャを 3 層アーキテクチャに抽象化します。リクエストフローのパス全体は次のとおりです。

1) クライアントがリクエストを送信します。 Kafka サーバーには Acceptor スレッドが存在します。送信された要求を監視するために、OP_ACCEPT イベントがこのスレッドにバインドされます。以下には、セレクターに送信されたリクエストがあるかどうかを継続的に監視する while 無限ループがあります。リクエストリンクを受信すると、ソケットチャネルにカプセル化され、ソケットチャネルはネットワークアーキテクチャの最初のレイヤーに送信されます。

2) アーキテクチャの最初の層には、3 つの同一のプロセッサスレッドがあります。これらの各スレッドには、ラウンドロビン方式でソケットチャネルを格納する接続キューがあります。リクエストが増加し続けると、接続キューに多くのソケットチャネルが存在するようになります。このとき、ソケットチャネルは各セレクターに OP_READ イベントを登録します。上の図の最初のレイヤーにある 3 番目のプロセッサスレッドを参照してください。各スレッドには、各ソケットチャネルを移動する while ループもあります。イベントをリッスンした後、クライアントから送信されたリクエストを受信します。このとき、プロセッサスレッドはリクエスト (送信されるリクエストはバイナリです。前述のように、ネットワーク間の送信にはシリアル化が必要です) を解析し、それを解析して Request オブジェクトにカプセル化し、上の図に示すネットワークアーキテクチャの第 2 層に送信します。

3) 第 2 層アーキテクチャには、RequestQueue (リクエストキュー) と ResponseQueue (リターンキュー) の 2 つのキューがあります。リクエストは、バッファとして機能するリクエストキューに保存されます。ここで、ネットワークアーキテクチャの第 3 層に到達します。

4) 第 3 層アーキテクチャには、8 つのデフォルトの RequestHandler スレッドを持つ RequestHandler スレッドプールがあります。開始後、これらの 8 つのスレッドは、第 2 層の RequestQueue キューからリクエストを継続的に取得し、リクエスト本体のデータを解析し、組み込みツールクラスを通じてデータをディスクに書き込みます。

5) 書き込みが成功したら、クライアントに応答する必要があります。このとき、Response オブジェクトがカプセル化され、返された結果は第 2 層の ResponseQueue キューに格納されます。現時点では、デフォルトで 3 つの小さな応答キューがあり、その数は第 1 層アーキテクチャのプロセッサスレッドに対応しています。

6) この時点で、第 1 層のプロセッサスレッドの while ループが応答要求を走査します。トラバーサルが完了すると、セレクターに OP_WRITE イベントが登録され、応答要求がクライアントに送り返されます。

7) プロセス全体に関係するパラメーターは 2 つあります: num.network.threads = 3 と num.io.threads = 8。デフォルトのパラメーターが十分ではないと思われる場合は、これら 2 つのパラメーターを、num.network.threads = 9、num.io.threads = 32 (CPU の数と一致するように) など、最適化できます。各 RequestHandler スレッドは 2000QPS、2000 * 8 = 16,000 QPS を処理でき、拡張後は 64,000 QPS をサポートできます。拡張後、Kafka は 60,000 QPS をサポートできます。上記のアーキテクチャから、Kafka は高同時実行リクエストをサポートできることがわかります。