インタビュアー：カフカをご存知ですか？ Kafka ログセグメントの読み取りと書き込みの方法について説明します。

[[332223]]

私がこの記事を書いた理由は、以前のインタビューでインタビュアーからこの質問をされたからです。面接官はこう尋ねました。「Kafka に精通しているとおっしゃいましたが、ソース コードは読みましたか? では、Kafka ログ セグメントの読み取りと書き込みの方法を教えてください。」

私は心の中で静かに言いました。「しまった...ソースコードを少し読んだと言ったのに、少しじゃなかった。これについて言及するべきではなかった！」ではどうすればいいでしょうか？家に帰って通知を待つしかありません。

しかし、将来的に地位を取り戻すためには、ただ座って死を待つわけにはいきません。私たちは少しずつから数十億まで、一歩ずつ進んでいかなければなりません。今日は、ソース コード レベルで Kafka ログ セグメントを読み書きする方法について説明します。

Kafka のストレージ構造

ご存知のとおり、Kafka のトピックには複数のパーティションを含めることができます。パーティションは、実際には最小の読み取りおよび保存構造です。つまり、コンシューマーはトピックをサブスクライブしているように見えますが、実際にはトピックの下のパーティションからメッセージを取得します。メッセージを送信するプロデューサーの場合も同様です。

トピックとパーティションの関係

上図は全体的な論理関係を示しています。ディスク上の実際のコードにマップされた関係を次の図に示します。

各パーティションは、ディスク上のサブディレクトリである Log オブジェクトに対応します。サブディレクトリの下には、複数のログ セグメントのグループ、つまり複数のログ セグメントが存在します。ログ セグメントの各グループには、メッセージ ログ ファイル (log で終わる)、変位インデックス ファイル (index で終わる)、およびタイムスタンプ インデックス ファイル (timeindex で終わる) が含まれます。実際、.txnindex、.deleted など、他のサフィックスを持つファイルもあります。スペースが限られているため、ここでは説明しません。

以下はログの定義である

以下はログセグメントの定義です

indexIntervalBytes は、インデックスを作成する前に挿入されるメッセージの数として理解できます。このことから、Kafka のインデックスは実際にはスパース インデックスであり、インデックス ファイルがメモリを占有しすぎるのを防ぎ、より多くのインデックスをメモリに格納できることがわかります。対応する値はブローカー側のパラメータ log.index.interval.bytes で、デフォルトは 4 KB です。

インデックスを介してメッセージを検索する実際のプロセスは、最初にオフセットを介してインデックスが配置されているファイルを見つけ、次にバイナリ検索を介してターゲットに最も近いインデックスを見つけ、最後にメッセージ ファイルを順番に走査してターゲット ファイルを見つけることです。この操作の時間計算量は O(log2n)+O(m) です。ここで、n はインデックス ファイル内のインデックスの数、m はスパース性です。

これは空間と時間の交換であり、データ構造とアルゴリズムのバランスです。素晴らしいですね！

rollJitterMsについて話しましょう。これは実際には妨害値です。対応するパラメータは log.roll.jitter.ms です。これは実際にはログ セグメントのセグメンテーションに関係しています。 log.segment.bytes はログ セグメント ファイルのサイズを制御します。デフォルトは 1G です。つまり、ファイル ストレージが 1G を超えると、新しいファイルが書き込まれます。これは寸法としてのサイズに基づいています。時間に基づいてセグメント化されるパラメーター log.segment.ms もあります。

このパラメータを設定した後、パーティションが多数あり、このパラメータがグローバルであるため、多数のファイルを同時に分割する必要がある場合、ディスク IO はそれを処理できません。したがって、分割するには rollJitterMs を設定する必要があります。

Redis Cache の有効期限はどうですか?多数のキャッシュが同時に期限切れになり、キャッシュがデータベースに侵入するのを防ぐために、有効期限に乱数を追加します。ほら、知識は普遍的なものなんだ！

ログセグメントの書き込み

1. 現在のログ セグメントが空かどうかを判断します。空の場合は、後でログセグメントを分割するための基準として時間を記録します。

2. 変位値が適正であることを確認します。最後に、AbstractIndex.toRelative(..) メソッドが呼び出され、オフセットが 0 未満か、最大 int 値より大きいかが判断されます。

3. メッセージを追加すると、実際に FileChannel を通じてメッセージが書き込まれます。もちろん、メモリとページ キャッシュにのみ書き込まれます。ディスクをフラッシュするかどうかは構成によって異なります。

4. ログ セグメントの最大タイムスタンプと、最大タイムスタンプに対応する変位値を更新します。このタイムスタンプは、実際には定期的なログ削除の基準として使用されます。

5. 必要に応じてインデックスエントリを更新します (bytesSinceLastIndexEntry > indexIntervalBytes)

最後にフローチャート

メッセージ作成プロセス

ログセグメントの読み取り

1. 最初のメッセージのオフセットに従って、OffsetIndex を通じて対応するメッセージの物理的な位置とサイズを見つけます。

2. LogOffsetMetadata を取得します。メタデータには、メッセージのオフセット、メッセージが配置されているセグメントの開始オフセット、および物理的な場所が含まれます。

3. minOneMessage が true かどうかを判断します。その場合は、メッセージ サイズを返すように調整します。実際、単一のメッセージが maxSize より大きい場合に、コンシューマーが飢餓状態になるのを防ぐためにこれを返すことができます。

4. 最大 fetchSize を計算します。これは、(最大物理変位 - このメッセージの開始物理変位) と調整済み最大サイズ (前の操作の調整済み最大サイズがすでにメッセージの最小サイズであるため、これはよくわかりません) の最小値です。

5. FileRecordsのsliceメソッドを呼び出して、指定された場所から指定されたサイズのメッセージセットを読み取り、FetchDataInfoを構築して返します。

ここに別のフローチャートを示します。

メッセージの読み取りプロセス

まとめ

転んだところから立ち上がるべきですよね？これを実行すれば、次回面接官からこの質問をされても怖くなくなるでしょう。

ソースコードはまあまあです、ハハハハ（まずは勢いが必要です）

実際、これは Kafka ソースコードの氷山の一角に過ぎず、まだ長い道のりが残っています。 Kafka Broker は Scala で書かれていますが、言語は問題ではありません。読みにくくなく、コメントも豊富です。知らない文法に出会ったら、調べれば終わりです。

したがって、ソース コードから始めて、ソース コードから理解することを強くお勧めします。今回お話ししたappend関数とread関数は非常にコアな関数ですが、一見すると複雑なものではないので、「ソースコード」という言葉に怖気付かないでください。

ソースコードを読むことで、内部設計の原則をより深く理解し、コーディング スキルを向上させることができます (ソースコードを見て、自分もこのように書けることに気づくことがよくあります)。もちろん、システムアーキテクチャ機能もあります。

じゃあ、僕にとって一番大事なのは、自慢できることなんです（笑）。

メロドラマ

白氏は監視画面を見つめながら、「なぜ？ Kafka Broker の物理ディスク I/O 負荷が急に高くなったのはなぜ？」と考えていました。老白の頭に数本の髪の毛が逆立っているのが、彼をとても無力に見せている。

「log.segment.ms パラメータを設定しましたか? log.roll.jitter.ms を試してください。」老白が顔を上げると、私は背が高く堂々とした姿と光沢のある禿げ頭を残して、すでにオフィスから出て行っていました。

「ハゲて強くなった」