Kafka: サーバー上でメッセージがどのように保存され、読み取られるかを本当にご存知ですか?

序文

Kafka が何千万ものメッセージをどのように処理できるのか興味があるかもしれません。メッセージはパーティションにどのように保存されますか?この記事では、メッセージがどのように保存されるかを説明します。この記事では、主にメッセージの論理ストレージと物理ストレージという 2 つの観点から実装の原則を紹介します。

記事の概要

• パーティション、レプリカ、ログ、およびログセグメント間の関係。
• メッセージフロー分析を記述します。
• メッセージの消費とレプリカの同期プロセスの分析。

パーティション、レプリカ、ログ、ログセグメントの関係

3 つのブローカー、1 つのトピック、3 つのパーティション、および 2 つのレプリカを持つ Kafka クラスターがあると仮定します。パーティションの物理的な分布を次の図に示します。

上の図からわかるように、トピックは 3 つのパーティションで構成され、各パーティションはマスターとスレーブの 2 つのレプリカで構成されます。各パーティションのマスター レプリカとスレーブ レプリカは、異なるブローカーに分散されます。このことから、ブローカーがダウンした場合、マスター レプリカのみが外部の読み取りおよび書き込み要求を提供するため、他のブローカーに分散されたスレーブ レプリカをマスター レプリカとして設定できることがわかります。もちろん、最新の 2.x バージョンでは、スレーブ レプリカは外部読み取り要求を行うこともできます。システムの可用性を向上させるために、マスター レプリカとスレーブ レプリカを異なるブローカーに分散します。

パーティションの実際の物理ストレージはログ ファイルの形式で表示され、各ログ ファイルは複数の LogSegments で構成されます。 Kafka はなぜこのように設計されているのでしょうか?実のところ、その理由は非常に単純です。メッセージが継続的に書き込まれると、ログ ファイルは確実に大きくなります。管理の便宜上、Kafka は大きなファイルを 1 つずつ LogSegments に分割して管理します。各 LogSegment はデータ ファイルとインデックス ファイルで構成されます。データ ファイルは実際のメッセージ コンテンツを保存するために使用され、インデックス ファイルはメッセージ コンテンツの読み取りを高速化するために使用されます。

友人の中には、Kafka 自体はパーティション次元の順序でメッセージを消費し、ディスクは順番に読み取るときに非常に効率的であるため、インデックスを使用する必要がないと尋ねる人もいるかもしれません。実際、パーティション内のメッセージをランダムに使用するなどの特別なビジネス ニーズを満たすために、Kafka はまずインデックス ファイルを通じてメッセージの実際の保存場所をすばやく見つけ、それを処理することができます。

Partition、Replica、Log、LogSegment の関係をまとめてみましょう。メッセージはパーティション ディメンションに基づいて管理されます。システムの可用性を向上させるために、各パーティションに対応する数のレプリカ コピーを設定できます。通常、レプリカ コピーの数はトピックを作成するときに指定されます。パーティションとレプリカ コピーの実際の物理ストレージ形式は、ログ ファイルを通じて表示されます。メッセージの継続的な書き込みによるログ ファイルのサイズの継続的な増加を防ぐために、ログは 1 つずつ LogSegment ファイルに分割されます。

注: 同時に、書き込み可能としてマークされている LogSegment は各プライマリ パーティションに 1 つだけ存在します。 LogSegment ファイルのサイズが一定サイズを超えると (たとえば、ファイル サイズが 1G を超えると、これは HDFS に保存されているデータ ファイルと同様です。HDFS のデータ ファイルが 128M に達すると、新しいファイルに分割されてデータが保存されます)、新しい LogSegment が作成され、新しく書き込まれたメッセージを引き続き受信します。

メッセージプロセス分析の書き込み

プロセス分析

3 番目の記事で述べたように、プロデューサー クライアントは、各パーティションごとに一度に一連のメッセージをサーバーに送信します。一連のメッセージを受信した後、サーバーはそれらを対応するパーティションに書き込みます。上記のプロセスは主に次のステップに分かれています。

• クライアント メッセージ コレクターは、同じパーティションに属するメッセージを収集し、各メッセージのオフセットを設定します。各メッセージ バッチは常に 0 から始まり、単調に増加します。たとえば、3 つのメッセージが初めて送信される場合、3 つのメッセージには順番に [0,1,2] という番号が付けられます。 2 回目に 4 つのメッセージが送信される場合、メッセージには順番に [0,1,2,3] の番号が付けられます。ここで設定されるメッセージ オフセットは相対オフセットであることに注意してください。
• クライアントはサーバーにメッセージを送信し、サーバーは次のメッセージの絶対オフセットを取得し、サーバーに送信されたメッセージのバッチの相対オフセットを絶対オフセットに変更します。
• 変更されたメッセージを現在アクティブな LogSegment に追加形式で追加し、絶対オフセットを更新します。
• メッセージ セットをファイル チャネルに書き込みます。
• ファイル チャネルはメッセージ セットをディスクにフラッシュし、メッセージの書き込み操作を完了します。

上記のプロセスを理解した上で、メッセージの具体的な構成を見てみましょう。

メッセージは次の 3 つの部分で構成されます。

• OffSet:オフセット。クライアントによってメッセージが送信される前に、相対オフセットがこの位置に保存されます。メッセージは、LogSegment に保存される前に、まず絶対オフセットに変更されてからディスクに書き込まれます。
• サイズ:このメッセージのコンテンツサイズ
• メッセージ:メッセージの具体的な内容。7 つの部分で構成されます。 CRC はメッセージの検証に使用され、属性は属性を表し、キー長と値長はそれぞれキーと値の長さを表し、キーと値はそれぞれ対応するコンテンツを表します。

メッセージオフセット計算プロセス

上記のプロセスから、各メッセージは実際にディスクに保存される前に絶対オフセットが割り当てられることがわかります。同じパーティション内では、メッセージの絶対オフセットは 0 から始まり、単調に増加します。異なるパーティションでは、メッセージの絶対オフセットに関係はありません。次に、メッセージの絶対オフセットの計算規則について説明します。

メッセージのオフセットを決定する方法は 2 つあります。 1 つは、各メッセージを順番に読み取って判断することです。この方法は比較的高価です。実際、私たちが知りたいのはメッセージの内容ではなく、メッセージのオフセットだけです。 2 番目の方法は、各メッセージのサイズ属性を読み取り、次のメッセージの開始オフセットを計算することです。たとえば、最初のメッセージの内容が「abc」の場合、ディスクに書き込んだ後のオフセットは、8 (オフセット) + 4 (メッセージ サイズ) + 3 (メッセージ コンテンツの長さ) = 15 になります。書き込まれた 2 番目のメッセージは「defg」で、その開始オフセットは 15 です。次のメッセージの開始オフセットは、15+8+4+4=31 になります。

消費メッセージとレプリカ同期プロセスの分析

メッセージの書き込みプロセスとは異なり、メッセージの読み取りプロセスは、コンシューマー側でメッセージを消費する場合と、レプリカ (バックアップ レプリカ) からプライマリ レプリカにメッセージを同期する場合の 2 つのケースに分かれています。読み取りプロセスの分析を開始する前に、使用されるいくつかの変数を理解する必要があります。そうしないと、プロセス分析が混乱する可能性があります。

• BaseOffSet:基本オフセット。各パーティションは N 個の LogSegments で構成されます。各 LogSegment には基本オフセットがあり、おおよそ次のものから構成されます。配列内の各数字は、LogSegment のベース オフセットを表します: [0、200、400、600、...]。
• StartOffSet:開始オフセット。コンシューマーがメッセージの読み取り要求を開始すると、メッセージの消費を開始する位置が指定されます。
• MaxLength:プルサイズ。コンシューマーがメッセージの読み取り要求を開始すると、取得するメッセージ コンテンツの最大データ サイズが指定されます。このパラメータは max.partition.fetch.bytes で指定でき、デフォルトのサイズは 1M です。
• MaxOffSet:最大オフセット。コンシューマーがメッセージをプルする場合、プルできるメッセージの最高位置は、一般に「最高水準点」と呼ばれます。このパラメータは、プロデューサーによってまだ書き込まれていないメッセージをコンシューマーが消費するのを防ぐためにサーバーによって指定されます。このパラメータは、スレーブ レプリカをマスター レプリカと同期する場合には使用されません。
• MaxPosition: LogSegment の最大位置。特定の LogSegment からの開始オフセットを決定します。 MaxLength を読み取った後、MaxPosition を超えることはできません。 MaxPosition はオフセットではなく、実際の物理的な位置です。

コンシューマーが位置 000000621 からメッセージを消費し始めると仮定すると、いくつかの変数間の関係は次の図に示されます。

レプリカを消費してプルするプロセスは次のとおりです。

• クライアントはプル場所、つまり StartOffSet の値を決定し、プライマリ コピーに対応する LogSegment を見つけます。
• LogSegment はインデックス ファイルとデータ ファイルで構成されます。インデックス ファイルは小さいものから大きいものの順に並べられているため、まずインデックス ファイルから StartOffSet 以下の最も近いインデックス位置を決定します。
• インデックスの場所に応じて対応するデータ ファイルの場所を見つけます。データ ファイルも小さいものから大きいものの順に並べられているため、見つかったデータ ファイルの場所から、メッセージを消費またはプルする場所である StartOffSet と等しい位置が見つかるまで逆方向に移動してください。

StartOffSet から MaxLength のデータを引き出してコンシューマーに返すか、レプリカから消費またはバックアップします。

プルされたメッセージの開始位置が 00000313 であると仮定すると、メッセージ プルのフロー チャートは次のようになります。

要約する

この記事では、論理ストレージと物理ストレージの観点からメッセージの書き込みと消費のプロセスを分析します。論理ストレージはパーティションを使用してメッセージのバッチを管理します。パーティション マップ ログ オブジェクト。ログ オブジェクトは複数の LogSegments を管理します。複数のパーティションが完全なトピックを構成します。メッセージの実際の物理的な保存は、LogSegment から 1 つずつ構成され、各 LogSegment はインデックス ファイルとデータ ファイルで構成されます。