カフカはなぜ止められないほど速いのでしょうか?

現在、市場には ActiveMQ、RabbitMQ、ZeroMQ など、ほとんどの人がすでによく知っている多くのメッセージ キューが存在しています。

[[327288]]
画像はPexelsより

特に、エンタープライズ バス通信における ActiveMQ の初期のアプリケーションと同様に、これは基本的にエンタープライズ レベルの IT 施設ソリューションに不可欠な部分です。

現在、Kafka は非常に安定しており、徐々に広く使用されるようになっています。もはや目新しいものではないが、カフカが雨後の筍のようにユニークで眩しい存在であることは否定できない。今日はカフカを注意深く分析し、その裏話を理解していきます。

次のコンテンツ バージョンは、最新の Kafka 安定バージョン 2.4.0 に基づいています。この記事には主に以下の内容が含まれています。

• Kafka が高速なのはなぜですか?
• Kafka が安定しているのはなぜですか?
• Kafkaの使い方

この記事はあくまでも入門ガイドです。後ほど、HBase、Spark、Kylin、Pulsar などの関連コンポーネントの分析が行われます。

Kafka が高速なのはなぜですか?

速いというのは相対的な概念です。比較しなければ害はありません。したがって、通常、Kafka は、IO を備え、主に IO に依存して情報を送信するメッセージ キューである一般的な ActiveMQ および RabbitMQ に関連していると言われています。

情報フローの送信にメモリのみに依存する ZeroMQ などのメッセージ キューは、もちろん高速になりますが、このようなメッセージ キューは特別なシナリオでのみ使用されるため、比較には含まれません。

したがって、Kakfa が高速であると言う場合、それは通常、次のシナリオに基づいています。

• スループット: 1 秒あたり数十万または数百万のメッセージを処理する必要がある場合、Kafka は他の MQ よりも高速に処理します。
• 高い同時実行性: 数百万または数千万の Consumer が存在する場合、Kafka は同じ構成のマシン上にさらに多くの Producer と Consumer を持つことになります。
• ディスク ロック: 他の MQ と比較して、Kafka では IO 操作を実行するときに IO を同期的にロックするシナリオが少なく、待機時間が短くなります。

上記の点を踏まえて、Kafka が高速である理由を詳しく見てみましょう。

メッセージキューのプッシュプルモデル

まず、単純にコンシューマーの観点から見ると、「Kafka は高速である」というのは誤った主張です。なぜなら、他の MQ と比較すると、プロデューサーがメッセージを生成してからコンシューマーがメッセージを消費するまでの時間は、他の MQ よりも長いか同等でなければならないからです。

その理由は、メッセージ キュー設計の 2 つのモデルに関係しています。

• プッシュモデル
• プルモデル

次の図に示すように:

プル モデルの場合、プロデューサーはメッセージを生成した後、それを MQ サーバーにアクティブに送信します。パフォーマンスを向上させてコストを削減するために、一部のクライアントはバッチで送信するように設計されています。

ただし、単一のメッセージであってもバッチであっても、プロデューサーはメッセージを MQ サーバーにアクティブにプッシュします。

MQ サーバーがメッセージを受信した場合、プル モデルでは、MQ サーバーはメッセージをコンシューマーにアクティブに送信せず、メッセージのオフセットを維持および記録することもありません。コンシューマーは、新しいメッセージが生成されたかどうかをサーバーに問い合わせるためのタイマーを自動的に設定します。

通常、クエリは 100 ミリ秒ごとに 1 回実行されます。新しいメッセージが生成されると、ローカル マシンに同期され、オフセットが変更されて記録されます。サーバーはオフセットの保存を支援できますが、オフセットの合理性を積極的に記録および検証することはありません。

同時に、コンシューマーはオフセットを完全に独立して維持し、カスタマイズされた情報の読み取りを実装できます。

プッシュ モデルの場合、サーバーはメッセージを受信した後、まずメッセージ情報を記録し、独自のメタデータ データベースから対応するメッセージ コンシューマーを照会します。

サーバーとコンシューマーは接続時に長いリンクを確立するため、メッセージはコンシューマーに直接送信できます。

Kafka はプル モデルに基づくメッセージ キューです。したがって、メッセージを取得する Consumer の観点から見ると、遅延はポーリング サイクル以下になるため、プッシュ モデルのメッセージ キューよりもメッセージ取得の遅延が大きくなりますが、プッシュ モデルにも問題があります。

まず、サーバーは、メッセージの送信先やオフセットなど、対応するコンシューマーのメタデータを記録する必要があり、また、メッセージをコンシューマーにプッシュする必要があるため、一連の問題が必然的に発生します。

現時点でネットワークが良好でなく、コンシューマーがメッセージを受信できず、メッセージが正常に送信されない場合はどうなるでしょうか?メッセージが送信されたと仮定した場合、それが受信されたかどうかをどのように確認すればよいでしょうか?

したがって、少なくとも 1 回の使用、および 1 回のみの使用などの特性を実現するには、サーバーとコンシューマーの間で複数層のパスワード確認が必要です。

Kafka などのプル モデルでは、この機能を Consumer に任せて自動的にメンテナンスするため、サーバーは不要な費用をさらに削減します。

したがって、同等のリソースの観点から見ると、Kafka にはより多くのリンクされたプロデューサーとコンシューマーが存在することになり、メッセージの輻輳が大幅に軽減され、より高速に見えるようになります。

OS ページ キャッシュとバッファ キャッシュ

太陽の下には新しいものは何もない。フレームワークの場合、実行速度を速める方法は通常いくつかしかありません。 Kafka はこの方法を極限まで活用しています。

その 1 つは、OS キャッシュ (主にページ キャッシュとバッファー キャッシュ) の使用を最大限に活用することです。

Linux を使用する学生は通常、これら 2 つのキャッシュに精通しています。たとえば、Linux で free コマンドを実行すると、次の出力が表示されます。

インターネットからの写真

buffers と cached という名前の 2 つの列と、「-/+ buffers/cache」という名前の行があります。これら 2 つの情報の具体的な説明は次のとおりです。

ページキャッシュ: ファイル システム レベルのキャッシュ。ディスクから読み取られたコンテンツはここに保存されるため、プログラムはディスクのコンテンツを非常に高速に読み取ることができます。たとえば、grep や find などの Linux コマンドを使用してコンテンツやファイルを検索する場合、1 回目は非常に遅くなりますが、2 回目は何倍も速くなり、ほぼ瞬時に検索されます。

さらに、ページ キャッシュ内のデータが変更された後、つまりダーティ データがディスクに書き込まれるとき、そのデータはディスクに直接書き込まれるのではなく、バッファー キャッシュに転送されます。

表示されるキャッシュ列の値は、現在のページ キャッシュの使用状況を示します。ページ キャッシュはファイルのページ データです。ページは論理的な概念なので、ページ キャッシュはファイル システムと同じレベルにあります。

バッファ キャッシュ: ディスクなどのブロック デバイスのバッファ。メモリのこの部分はディスクに書き込まれます。

バッファ列には、現在のバッファ キャッシュの使用状況が表示されます。バッファ キャッシュは、ブロック デバイス (ディスクなど) のブロック データをキャッシュするために使用されます。ブロックは物理的な概念であるため、バッファ キャッシュはブロック デバイス ドライバーと同じレベルにあります。

どちらもデータ IO を高速化し、書き込まれたページをダーティとしてマークし、外部ストレージにフラッシュするために使用されます。データを読み取るときは、まずキャッシュが読み取られます。失敗した場合は、外部ストレージに読み込まれ、読み取られたデータもキャッシュに追加されます。

オペレーティング システムは、すべての空きメモリを常にページ キャッシュおよびバッファー キャッシュとしてアクティブに使用します。 OS のメモリが不足すると、LRU などのアルゴリズムを使用してキャッシュ ページが削除されます。

上記の概念を踏まえて、Kafka がこの機能をどのように使用するかを見てみましょう。

まず、データ IO の場合、通常は次のプロセスが発生します。

• オペレーティング システムは、ディスクからカーネル領域のページ キャッシュにデータをコピーします。
• ユーザープログラムはカーネル領域のページキャッシュをユーザー領域キャッシュにコピーします。
• ユーザープログラムは、ユーザー領域内のバッファをソケットバッファにコピーします。
• オペレーティング システムは、ソケット キャッシュ内のデータをネットワーク カードのバッファーにコピーし、データを送信します。

1 回の IO 要求操作で 2 回のコンテキスト スイッチと 4 回のシステム コールが実行され、同じデータがキャッシュに複数回コピーされることがわかります。実際、コピーはカーネル状態で直接実行できます。

つまり、2番目と3番目の手順を省略すると、次のようになります。

このようにデータを変更するプロセスのため、Kafka は最初に設計されたときにこのプロセスを参照し、OS ページ キャッシュを最大限に活用してデータをコピーし、ディスク操作を最小限に抑えました。

Kafka の生成と消費がうまく連携すると、データは完全にメモリに保存され、クラスターのスループットが大幅に向上します。

初期のオペレーティング システムでは、ページ キャッシュとバッファー キャッシュは 2 つの別々のキャッシュでした。その後、同じデータが 2 回キャッシュされる可能性があることが判明したため、ほとんどの場合、2 つが 1 つに結合されました。

Kafka は JVM 言語で記述されており、動作中に JVM および JVM GC から切り離すことはできませんが、Kafka 自体はキャッシュを管理せず、OS ページ キャッシュを直接キャッシュとして使用します。

これにより、次のような利点がもたらされます。

• JVM 内のすべてはオブジェクトであるため、オブジェクトのサイズに関係なく、スペースを浪費する余分な JVM オブジェクト メタデータが常に存在します。
• JVM 自体の GC はプログラムによって手動で制御されないため、JVM をキャッシュとして使用すると、大きなオブジェクトや頻繁な GC に遭遇したときにシステム全体のスループットが低下します。
• プログラムが異常終了または再起動すると、すべてのキャッシュが無効になり、災害復旧アーキテクチャにおける迅速な復旧に影響します。ページ キャッシュは OS キャッシュであるため、プログラムが終了しても存在します。

そのため、Kafka は IO プロセスを最適化し、ページ キャッシュを最大限に活用します。これにより、時間の消費が少なくなり、スループットが高くなり、他の MQ よりも高速になります。

次の図は、3 つの関係を簡単に表したものです。

プロデューサーとコンシューマーのレートが似ている場合、Kafka は情報をディスクに書き込むことなくほぼ完全に送信できます。

追加シーケンシャル書き込み

上記の重要な機能に加えて、Kafka には、データの永続的な保存に順次追加書き込みを使用する設計もあります。 Kafka は、各トピックのパーティション ファイルにメッセージをドロップするときに、ディスクの高速シーケンシャル アクセス機能を最大限に活用して、メッセージを順番に追加します。

インターネットからの写真

Kafka のファイルは、トピックの下のパーティションに従って保存されます。各パーティションには独自のシーケンス ファイルがあります。各パーティションのシーケンスは共有されません。主な分割は、メッセージ キーのハッシュに基づいて、どのパーティションに該当するかを決定します。

まず、Kafka の特徴を完全に理解できるように、Kafka のさまざまな用語を詳しく説明しましょう。

• ブローカー: Kafka でメッセージを処理するために使用されるサーバー。これは Kafka クラスター内のノードでもあります。複数のノードが Kafka クラスターを形成します。
• トピック: メッセージのトピック。各ビジネス システムまたはコンシューマーは、メッセージを取得するために 1 つ以上のトピックをサブスクライブする必要があります。プロデューサーは、情報送信のパスワード名と同じ、メッセージを送信するトピックを指定する必要があります。
• パーティション: トピックは複数のパーティションに分割され、ディスクに保存されます。ファイルは、Kafka によって構成されたストレージ ディレクトリ内の対応するパーティション ID に従って作成されたフォルダーに保存されます。ディスク上で表示される最大のストレージ ユニットです。
• セグメント: パーティションには、実際にコンテンツを保存するための複数のセグメント ファイルがあります。
• オフセット: 各パーティションには独自の独立したシーケンス番号があり、その範囲は現在のパーティションに限定され、対応するファイルの内容を読み取るために使用されます。
• リーダー: 各トピックには、トピックに情報を書き込み、データの一貫性を維持する責任を負うリーダーが必要です。
• コントローラー: 各 Kafka クラスターは、コントローラーとして機能するブローカーを選択します。コントローラーは、各トピックのリーダーを決定し、クラスター ブローカー情報の変更を監視し、クラスターの正常性を維持する役割を担います。

ディスク上に最終的に保存されるのはセグメント ファイルであることがわかります。各パーティション (ディレクトリ) は、同じサイズの複数のセグメント データ ファイルに均等に分散された巨大なファイルに相当します。

ただし、各セグメント ファイル内のメッセージの数は必ずしも同じではありません。この機能を使用すると、古いセグメント ファイルをすばやく簡単に削除できます。

Kafka はパーティションに基づいてメッセージを処理するため、パーティションに対応する順序を維持するだけでよく、セグメントは独立してステータスを維持できます。

セグメント ファイルは、インデックス ファイルとデータ ファイルで構成されます。ディスク上のファイルのサフィックスは .index と .log です。ファイルは、以下に示すように、シーケンス番号に従って生成されます。

インターネットからの写真

インデックスはデータの物理アドレスを維持し、データはデータのオフセット アドレスを格納します。それらは互いに関連しています。これはディスクの順次書き込みとはほとんど関係がないようです。 HDFS のブロック ストレージについて考えてみましょう。毎回固定サイズのブロックが要求され、ここでのセグメントは非常に似ています。

また、インデックス ドキュメント自体もファイル名に Offset が付けられているため、検索時に検索したい Offset に応じて該当するファイルをすばやく見つけ、そのファイルに基づいてコンテンツを取得できます。

そのため、Kafka の検索プロセスは、まず検索対象のオフセットに従ってファイル名をバイナリ検索し、該当するファイルを見つけ、次にインデックス メタデータの物理アドレスとログ ファイルのオフセット位置に基づいた順序で該当するオフセット位置の内容を読み取ります。

セグメント インデックス ファイルはスパース インデックス ストレージを採用しており、インデックス ファイルのサイズが削減され、mmap を通じてメモリを直接操作できます。スパース インデックスは、データ ファイルの対応する各メッセージのメタデータ ポインターを設定します。

高密度インデックスよりも多くのストレージスペースを節約できますが、検索に時間がかかり、特にランダム読み取りのシナリオでは、Kafka はあまり適していません。そのため、Kafka の特別なストレージ設計により、Kafka もより高速に感じられます。

Kafka が安定しているのはなぜですか?

Kafka が高速である理由については先ほど説明しました。 Kafka には高速機能に加えて、安定性という別の機能もあります。

Kafka の安定性はいくつかの側面に反映されています。

• データは安全であり、データが失われる可能性はほとんどありません。
• クラスターは安全であり、消費者は障害を意識することなく別のクラスターに切り替えることができます。
• 高可用性: 一部のパーティションが使用できない場合でも、残りのパーティションのデータは影響を受けません。
• フロー制御制限により、多数の消費者がサーバーの帯域幅を低下させることを防ぎます。

電流制限機構

Kafka の安定性は通常、全体的なアーキテクチャ設計によって決まります。多くの優れた機能が組み合わさることで、さらに優れたものになります。カフカの「クトータ」もその一つです。

フロー制限であるため、コンシューマーまたはプロデューサーのトラフィック帯域幅を制御する必要があることを意味します。通常、フロー制限は、一般的な N ウェイ スイッチやハイエンド ルーターなどのネットワーク カードで処理する必要があります。

したがって、Kafka が OS ネットワーク カードを操作してトラフィックを制御するのは明らかに非常に困難であるため、Kafka は別の特別なアイデアを採用しています。

つまり、ネットワーク カードを通過するトラフィックの量を制御する方法はありませんが、データが返される時間を制御する方法があります。 JVM プログラムの場合、それは単に Wait または Seelp の問題です。

したがって、Kafka には特別な遅延計算ルールのセットがあります。 Kafka は、ウィンドウに従って単位時間あたりに送信されたトラフィックをカウントします。

トラフィック サイズが設定されたしきい値を超えると、フロー制御がトリガーされ、現在のリクエストが Kafka の Qutota マネージャーにスローされます。遅延時間に達すると、データが再度返されます。

Kafka の ClientQutotaManager クラスのメソッドを見てみましょう。

これらのコード行は、Kafka の現在の制限計算ロジックを表しています。一般的な考え方は、現在のフロー制限を T 以下に設定すると、ウィンドウに基づいて計算される現在のレートは O になるというものです。

O が T を超えると速度制限が実施され、速度制限は次のようにアナウンスされます。

 X = (O - T)/ T * W

Xは遅延に必要な時間です。例を挙げてみましょう。トラフィックを 10MB/秒以下に制限し、過去 5 秒間のトラフィック (パブリック ウィンドウ間隔の W) が 100MB であるとすると、遅延時間は (100-5*10)/10=5 秒になります。

これにより、次のウィンドウが完了した後、フロー全体のサイズが制限を超えないことが保証されます。

KafkaApis の Producer と Consumer のコールバック コードを通じて、現在の制限の遅延戻りを確認できます。

Kafka の現在の制限については、デフォルトではクライアント ID またはユーザーに応じてフローが制限されます。実用の観点から見ると、これはあまり意味がありません。トピックまたはパーティションのパーティション レベルに基づく現在の制限には、より幅広い使用シナリオがあります。

選挙の仕組み

Kafka の背後にあるメタ情報は Zookeeper に大きく依存しています。ここでも、Zookeeper 自体については説明せず、Kafka が ZK をどのように使用するかに焦点を当てます。

まず、図で Kafka が ZK に大きく依存していることがわかります。

写真はインターネットから

ZK を使用する上で最も重要なことは、独自の情報の保存に加えて、Kafka が ZK を使用して選出メカニズムを実装し、その中でコントローラーが主な導入となることです。

まず、コントローラーは Kafka の中核であり、主に以下の重要な事項を担当します。ただし、これに限定されません。

つまり、コントローラーは Kafka の中心的な役割です。コントローラーには公正な競争が採用されており、どのブローカーもコントローラーになることができるため、クラスターの堅牢性が保証されます。

コントローラーの選出プロセスは次のとおりです。

① まずZK/Controllerノードの情報とControllerのブローカーIDを取得します。ノードが存在しない場合（たとえば、クラスターを作成したばかりの場合）、取得されるコントローラー ID は -1 になります。

②コントローラIDが-1でない場合、つまりコントローラがすでに存在する場合は、そのまま処理が終了します。

③コントローラーIDが-1の場合、コントローラーがまだ存在しないことを意味します。この時点で、現在のブローカーはコントローラーを ZK に登録し始めます。

④登録が成功すると、現在のブローカーがコントローラーになります。この時点で、onBecomingLeader() メソッドが呼び出され、コントローラーが正式に初期化されます。

注: コントローラー ノードは一時ノードです。現在のコントローラーが ZK セッションから切断されると、コントローラーの一時ノードが消え、コントローラーの再選出がトリガーされます。

⑤登録に失敗した場合（コントローラーが他のブローカーによって作成された場合、例外がスローされた場合など）、そのまま戻ります。

そのコードは KafkaController を通じて直接確認できます。

コントローラーが選出されると、他のブローカーは ZK の変更を監視し、クラスター内のコントローラーの障害に対応します。

これにより、新しいコントローラー選択アクションがトリガーされます。 Kafka の場合、全体の設計は非常にコンパクトで、コードの品質は非常に高く、多くの設計も非常に有益です。同様の機能は Kafka の多くの機能に反映されています。これらの機能が組み合わされて、Kafka の全体的な安定した状況が形成されます。

Kafkaの使い方

Kafka は全体的に非常に優れているように見えますが、万能の万能薬ではなく、それに応じた欠点があるはずです。したがって、Kafka がどのようなものか、またはそれをより効果的に使用する方法を理解するには、実際に練習して感覚をつかむ必要があります。

まとめてみると、次のようなさまざまな使用シナリオと特性がわかります。

①Kafkaは高頻度取引システムには適していない

Kafka は非常に高いスループットとパフォーマンスを備えていますが、単一メッセージの低レイテンシという点では、Kafka が従来の MQ ほど優れていないことは否定できません。結局のところ、プッシュ モデルに基づく MQ は、リアルタイム メッセージ送信のシナリオで固有の利点を得ることができます。

②Kafkaには完全なトランザクション機構がない

0.11 以降、Kafka はプロデューサーのバッチ送信を保証するトランザクション メカニズムを追加しました。ダーティ データが読み取られないようにするために、コンシューマーはメッセージ ステータスをフィルター処理して不適切なデータを除外できますが、すべてのデータを読み取る操作は保持されます。

それでも、Kafka のトランザクション メカニズムはまだ不完全です。その主な理由は、Kafka はクライアントに関心がないため、すべての共通プロトコルを統一しないからです。したがって、一度だけ消費するようなシナリオでは、効果はクライアントの実装に大きく依存します。

③Kafkaのリモート災害復旧ソリューションは非常に複雑です

Kafka の場合、データセンター間でシームレスな切り替えを実現するには、クラスター間プロキシをサポートする必要があります。

Kafka の特別な追加ログ設計メカニズムにより、同じオフセットを異なるブローカーや異なるコンテンツで再利用することはできません。

つまり、ファイルが別のサーバーにコピーされると、読み取れなくなります。データベースベースの MQ と比較すると、クラスター間でデータを同期することが困難です。

同時に、オフセットを再現することも非常に困難です。私たちはかつて、データセンター全体に Kafka クラスター プロキシを実装する顧客を支援しましたが、これには多額の費用がかかりました。

④Kafkaコントローラーアーキテクチャではクラスターリソースを十分に活用できない

Kafka コントローラーは、ES の分散化コンセプトに似ています。選出ルールに従って、クラスターからサーバーをコントローラーとして選択します。

これは、サーバーがコントローラーとブローカーの両方の責任を負うことを意味します。その結果、大量のメッセージの圧力により、サーバーのリソースがクラスターのボトルネックになりやすくなり、クラスター リソースを最大限に活用できなくなります。

コントローラーは HA をサポートしていますが、分散はサポートしていません。つまり、Kafka のパフォーマンスを最適化するには、各サーバーが少なくとも最高の構成に到達している必要があります。

⑤Kafkaにはあまりインテリジェントなパーティションバランス機能がない

通常、オンサイト ストレージを設計する場合、ホット スポットや十分に高いパフォーマンスが求められるシナリオでは、SSD と HD の組み合わせが使用されます。

同時に、クラスターのディスク容量が不均一な場合、Kafka にとって非常に深刻な問題になります。 Kafka パーティションはパーティションの数に基づいて生成され、次の図に示すように、ディスク数が最も少ないディスク上に新しいパーティションが作成されます。

私はかつて、ある会社が容量を考慮してパーティション作成ルールを変更するのを手伝いました。つまり、ディスク容量に応じてパーティションを選択するのです。

これにより、2 番目の問題が発生します。大容量のディスクにはパーティションが多く存在するため、ディスクに大量の IO が圧迫されることになります。最終的に、問題は IO に戻り、ディスク上の他のトピックのパフォーマンスに影響します。

そのため、MQ システムを検討する場合は、Kafka のパーティショニング ルールを手動で合理的に設定する必要があります。

結論

Kafka が唯一の解決策ではありません。数年前に勢いよく登場した新しいフレームワークである Pulsar は、Kafka に代わるというスローガンを掲げて市場に参入しました。これは、Kafka の問題点のいくつかを解決する次のフレームワークになる可能性があります。 Pulsarについては以下で説明します。

著者: バイファチュアン

編集者：タオ・ジアロン

出典：WeChatパブリックアカウントThoughtWorks Insight（ID：TW-Insight）から転載