ブローカーの実装ロジック - Kafka ナレッジ システム (パート 3)

[[409670]]

前回の記事では、Kafka プロダクション側のロジックと、メッセージがキャッシュに送信され、送信スレッドによってブローカーに送信される方法について説明しました。では、ブローカーはどのようにしてデータを受信して​​保持するのでしょうか?ブローカーのネットワーク設計から始めましょう。

ブローカーネットワーク設計

Kafka のネットワーク設計は Kafka のチューニングに関連しているため、高い同時実行性をサポートできます。

Kafka の 3 層ネットワーク アーキテクチャ

まず、すべてのクライアント要求が Acceptor に送信されます。ブローカーには 3 つのスレッドが存在します (デフォルトは 3)。これら 3 つのスレッドはプロセッサと呼ばれます。

アクセプターはクライアントのリクエストを一切処理しません。これを socketChannel に直接カプセル化し、これらのプロセッサに送信してキューを形成します。送信方法はポーリングです。つまり、最初に最初のプロセッサに送信し、次に 2 番目、3 番目に送信し、最後に最初のプロセッサに戻します。

コンシューマ スレッドがこれらの socketChannels を消費すると、リクエストが 1 つずつ取得され、これらのリクエストにはデータが付随します。

デフォルトでは、スレッド プールには 8 つのスレッドがあります。これらのスレッドは、リクエストの処理、リクエストの解析、およびリクエストが書き込みリクエストである場合はディスクへの書き込みに使用されます。読み取られた場合、結果が返されます。プロセッサは応答から応答データを読み取り、それをクライアントに返します。これは Kafka の 3 層ネットワーク アーキテクチャです。

調整ポイント1

したがって、Kafka を強化および調整する必要がある場合は、プロセッサを追加し、スレッド プール内の処理スレッドの数を増やすことで、目的の効果を実現できます。プロセッサがリクエストをあまりにも速く生成し、それを時間内に処理するのに十分なスレッドがないという問題を考慮して、リクエスト部分とレスポンス部分は実際にはキャッシュとして機能します。これは、リアクター ネットワーク スレッド モデルの拡張バージョンです。

ブローカーデータストレージ設計

データファイルをパーティション分割する

トピックは論理的な概念であり、パーティションはトピックの物理的なグループ化であり、トピックは複数のパーティションに分割でき、各パーティションは順序付けられたキューであることがわかっています。

たとえば、report_pushとlaunch_infoという2つのトピックを作成し、パーティションの数を4に設定します。ストレージパスとディレクトリのルールは次のとおりです: xxx/message-folder

 | --report_push-0  
 | --レポートプッシュ-1  
 | --レポートプッシュ2  
 | --レポートプッシュ3  
 | --launch_info-0  
 | --launch_info-1  
 | --launch_info-2  
 | --launch_info-3  

パーティションは物理的に複数のセグメントで構成されます。

【セグメント】ログ

各セグメントは同じサイズで、順番に読み書きされます。

各セグメントデータファイルはセグメント内の最小オフセットにちなんで命名され、ファイル拡張子は.logです。

ログのロールバックは log.segment.bytes によって制御され、デフォルト値は 1G です。

このように、指定されたオフセットを持つメッセージを検索する場合、バイナリ検索 (スキップ テーブル) を使用して、メッセージが配置されているセグメント データ ファイルを見つけることができます。

ディスク上では、パーティションはディレクトリであり、各セグメントはインデックス ファイルとログ ファイルで構成されます。次のように：

 $ ツリー カフカ |ヘッド-n 6
カフカ
├── イベント-1
 │ ├── 00000000003064504069.インデックス 
 │ ├── 00000000003064504069.log
 │ ├── 00000000003065011416.インデックス 
 │ ├── 00000000003065011416.log

セグメントの下のログファイルはメッセージが保存される場所です

各メッセージには、メッセージ本文、オフセット、タイムスタンプ、キー、サイズ、圧縮コーデック、チェックサム、メッセージのバージョン番号などが含まれます。

ディスク上のデータは、プロデューサーがブローカーに送信するデータやコンシューマーが受信するデータとまったく同じ形式です。ディスク形式はコンシューマーとプロデューサーのデータ形式とまったく同じであるため、Kafka はゼロコピー技術を通じて転送効率を向上させることができます。

【セグメント】インデックス

インデックス ファイルはメモリにマップされます。

インデックス ファイルはスパース形式のインデックスであり、パラメーター log.index.interval.bytes によって制御され、デフォルトは 4 KB です。つまり、すべてのデータがインデックス化されるわけではありません。デフォルトでは、書き込まれるデータの 4 KB ごとにインデックスが書き込まれます。

Kafka は、セグメント化されたデータ ファイルごとにインデックス ファイルを作成します。ファイル名はデータファイル名と同じですが、ファイル拡張子は .index です。

インデックス ファイルは、データ ファイル内の各メッセージに対してインデックスを作成するのではなく、スパース ストレージ方式を使用して、一定数のデータごとにインデックスを作成します。

これにより、インデックス ファイルが多くのスペースを占有することがなくなり、インデックス ファイルをメモリ内に保持できるようになります。

メモリマッピングについて:

• ハードディスクにデータを順次書き込んでも、ハードディスクのアクセス速度がメモリに追いつきません。そのため、Kafka のデータはハードディスクにリアルタイムで書き込まれません。最新のオペレーティング システムのページング ストレージを最大限に活用してメモリを活用し、I/O 効率を向上させます。メモリマップファイル（以下、mmap と呼びます）もメモリマップファイルに変換されます。その動作原理は、オペレーティング システムのページを直接使用して、ファイルを物理メモリに直接マッピングすることです。マッピングが完了すると、物理メモリ上の操作がハードディスク (適切な場合はオペレーティング システム) に同期されます。 mmap を使用すると、プロセスはハードディスクの読み取りと書き込みと同じようにメモリの読み取りと書き込みを行います。仮想メモリによってメモリがカバーされるため、メモリのサイズを心配する必要はありません。 Mmap は実際にはメモリ マッピングを実装するために使用される Linux の機能です。 Java NIO では、MappedByteBuffer を使用してメモリ マッピングを実装できます。

[Kafka でオフセットを介してメッセージ コンテンツをクエリするプロセス全体]

Kafka には、各ログ セグメントを保存するための ConcurrentSkipListMap があります。

offset-->concurrentSkipListMap-->baseOffset に対応するログ セグメントを検索-->インデックス ファイルを読み取ります。index--> offset-baseoffset を超えない最大のインデックス項目を検索-->セグメント ファイル (.log) を読み取り-->ログ セグメント ファイル (.log) から順番に検索します。

現在のインデックス ファイルのファイル名は baseOffset の値です。

ログ保持戦略

Kafkaは定期的に古いメッセージを削除するかどうかをチェックします。パラメータを参照してください。

log.retention.check.interval.ms、デフォルトは 5 分です。現在、ログ保持戦略は 3 つあります。

スペースに基づく: log.retention.bytes、デフォルトでは有効になっていません。

時間ベース: log.retention.hours (分/ミリ秒)、デフォルトは 7 日。

開始変位に基づく: Kafka 0.11.0.0 で導入され、ストリーム処理シナリオで処理された中間メッセージを削除する問題を解決します。

現在、時間ベースのログ保持戦略が最も一般的に使用されています。

調整ポイント2

つまり、クライアントのバージョンとブローカーのバージョンを一致させるようにしてください。

つまり、クライアントのバージョンとブローカーのバージョンの一貫性を保つようにしてください。バージョン間の不整合を過小評価しないでください。そうしないと、Kafka はゼロ コピーなどの多くのパフォーマンス上の利点を失うことになります。

図中の青いプロデューサー、コンシューマー、ブローカーは同じバージョンであり、それらの間の通信はゼロコピーの高速チャネルを利用できます。逆に、低バージョンの Consumer プログラムが Producer および Broker と対話する場合、転送には JVM ヒープのみを使用するため、高速チャネルが失われ、低速チャネルを使用する必要があります。したがって、ブローカー層を最適化する場合、サーバーとクライアントのバージョンを一貫して維持するだけで、パフォーマンス上の大きなメリットが得られます。

ブローカーレプリケーションメカニズム

パーティションレプリカのデフォルト数は1です。パラメータを参照してください。

デフォルトのレプリケーション係数。

【レプリカの役割（読み書き分離は提供しない）】

1. 冗長性を実装し、メッセージの信頼性を向上させる

2. 高可用性を実現し、リーダー選出に参加し、リーダーが利用できない場合に可用性を向上させます。

3. リーダーはパーティションのすべての読み取りおよび書き込み要求を処理します。フォロワーはリーダーのデータを受動的かつ定期的にコピーする

【リーダーレプリカ選挙】

1. 管理者は責任を負う

2. 選挙の仕組みや戦略

すべてのレプリカは、総称して割り当てレプリカ、またはARと呼ばれます。

情報サービス

SR は AR のサブセットです。 ISR リストはリーダーによって管理され、フォロワーがリーダーからのデータを同期する際には若干の遅延が発生します。しきい値を超えるフォロワーは ISR から削除され、OSR (Out-of-Sync Replicas) リストに保存されます。新しいフォロワーも最初に OSR に保存されます。 AR = ISR + OSR

基本的な戦略は、ISR にある AR から最初に生き残ったレプリカを見つけることです。

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

[コピーメカニズムの利点]

一般的に言えば、コピー メカニズムの利点は次のとおりです。

1. データの冗長性を提供します。システムの一部のコンポーネントに障害が発生しても、システムは引き続き動作できるため、全体的な可用性とデータの耐久性が向上します。

2. 高いスケーラビリティを提供します。水平拡張をサポートしており、マシンを追加することで読み取りパフォーマンスを向上させ、読み取り操作のスループットを向上させることができます。

3. データの局所性を向上させる。データをユーザーの地理的な場所の近くに配置できるため、システムの遅延が短縮されます。

Apache Kafka の場合、現時点ではレプリカ メカニズムによってもたらされる最初の利点、つまりデータの冗長性を提供して高可用性と高耐久性を実現することしか享受できません。

クライアント ユーザーの場合、Kafka のフォロワー レプリカは効果がありません。 MySQL のようにリーダー レプリカを「読み取り耐性」にすることも、一部のレプリカをクライアントの近くに配置してデータの局所性を向上させることもできません。

ブローカーのハイウォーターマークメカニズム

【コンセプト】

HW (ハイ ウォーター マーク) は、Kafka レプリカ オブジェクトの重要な属性です。パーティションの最高水準点は、リーダー レプリカの最高水準点によって表され、フォロワー レプリカによって同期された後の位置を意味します。

リーダーが新たに書き込んだメッセージについては、消費者はそれをすぐに消費することはできません。リーダーは、メッセージが ISR 内のすべてのレプリカによって同期されるのを待機し、その後 HW を更新します。そうして初めて、メッセージは消費者によって消費されるようになります。

【効果】

メッセージの可視性を定義します。パーティションの最高水準点以下のメッセージのみを消費できます。

Kafka がレプリカの同期を完了するのに役立ちます。 Kafka は、ハイウォーターマークに基づく非同期レプリカ同期メカニズムです。

【LEOのコンセプト】

これは、ログ終了オフセット (Log End Offset)、つまり次のメッセージが書き込まれるオフセットを意味します。

まとめると、なぜ MySQL のインデックスは Kafka のインデックス メカニズムを使用しないのでしょうか?

Kafka は非常に優れていて高速なのに、なぜ MySQL は Kafka のインデックス メカニズムを使用しないのでしょうか?

また、InnoDB でインデックスを維持するコストが Kafka よりも高いという問題も考慮する必要があります。 Kafka では、ConcurrentSkipListMap はデータが書き込まれるたびに更新されるのではなく、新しいインデックス ファイルが作成されるときにのみ更新されます。この領域でのメンテナンス作業は基本的に無視できます。 B+ ツリーでデータが挿入、更新、または削除されると、インデックスを更新する必要があり、「ページ分割」などの比較的時間のかかる操作も発生します。 Kafka のインデックス ファイルも順番に追加されるファイルであるため、B+ ツリーよりもはるかに少ない作業で済みます。

実際、最終的にはさまざまなアプリケーション シナリオによって決まります。 MySQL では CRUD 操作を頻繁に実行する必要があります。 CRUD は MySQL の主な作業内容です。この操作をサポートするには、メンテナンス要件がはるかに大きい B+ ツリーが必要です。 Kafka 内のメッセージは通常、ディスクに順番に書き込まれ、その後ディスクから順番に読み取られます (ページ キャッシュなどについては詳しく説明しません)。主な作業内容は書き込み＋読み取りであり、検索クエリ操作はほとんどありません。つまり、検索クエリは Kafka の補助的な機能に過ぎず、この機能のために高レベルのインデックスを維持するために多大なコストをかける必要はありません。前述したように、Kafka におけるこのアプローチは、ディスク容量、メモリ容量、検索時間、およびその他の側面の間の妥協点です。