Kafka のこれらの原則を説明していただけますか?

もしあなたが一人の女性だけを追いかけ、彼女に特別な注意を払っているなら、それを知っている人は間違いなくあなたがいい男だと言うでしょう。ちょっとだけ試してみて、ほとんどの女の子を追いかけたら、知っている人は間違いなくあなたをクズ呼ばわりするでしょう。

[[287183]]

テクノロジーについても同じことが言えます。テクノロジーを深く探求すれば、あなたは間違いなくこの分野で欠かせない人材になるでしょう。あらゆる技術を学んで困難に遭遇したら撤退するのであれば、すぐに代替されるでしょう。

現在の中国社会は、速いペースで人々を生み出す触媒のようなものです。週末にこんなことを聞きました。私の大学にも、そんな人がいて、クラス委員をしていたんです。彼の家族はある程度お金持ちなので、社会に早くから触れ、行動も成熟しているのかもしれません。彼はカウンセラーと非常に良い関係を築いています。彼が頻繁に授業を欠席しても、カウンセラーが欠席届の記入を手伝ってくれるほど、彼とカウンセラーの関係は良好だと言えます。彼の成績は非常に良かったので、クラスメイトがクラスモニターを評価すると、カウンセラーがこっそりと彼のスコアを 60 から 80 に変えました。私はこう言いました。「これは社会的触媒作用の産物だ」キャンパスも小さな社会であり、この小さな社会の中で多くのことが悪化し始めました。

私たちは、入手したものが準拠していて合法であるかどうかについて決して考えません。私の人生の指導者の一人がかつて私にこう言ったのを覚えています。「良い人間になりなさい。そして、いつも近道を探さないでください。」この文章はお茶と同じで、ゆっくり味わうことで初めて美味しくなります。

さて、テクノロジーの話に戻りましょう。

Kafka アプリケーションを開発しているだけの場合、または実稼働環境で Kafka を使用しているだけの場合は、Kafka の内部の仕組みを理解する必要はありません。ただし、Kafka の内部の仕組みを理解することは、Kafka の動作を理解するのに役立ち、問題を迅速に診断するのにも役立ちます。これら 3 つの問題について検討してみましょう。

Kafkaの複製方法
Kafka がプロデューサーとコンシューマーからのリクエストを処理する方法
Kafka のストレージの詳細は何ですか?

ご興味がございましたら、ぜひ時間を取ってこの記事をじっくりお読みください。

クラスターメンバー間の関係

Kafka は ZooKeeper 上で実行されることがわかっています。 ZooKeeper はクラスター形式で表示されるため、Kafka もクラスター形式で表示されます。これには、複数の生産者と複数の消費者をどのように調整するかという問題も含まれます。クラスター間の関係の維持も ZooKeeper によって実行されます。 Kafka クラスター間には複数のホスト (ブローカー) が存在し、各ブローカーには broker.id があることを知っておく必要があります。各 broker.id には、それを区別するための一意の識別子があります。この識別子は、構成ファイルで手動で指定することも、自動的に生成することもできます。

Kafka は、broker.id.generation.enable と reserved.broker.max.id を通じて新しい broker.id を生成できます。

broker.id.generation.enable パラメータは、broker.id を自動生成する機能を有効にするかどうかを構成するために使用されます。デフォルトでは true になっており、この機能が有効になっていることを意味します。自動的に生成される broker.id のデフォルト値は 1000 です。つまり、自動的に生成される broker.id はデフォルトで 1001 から始まります。

Kafka が起動すると、ZooKeeper の /brokers/ids パスに現在のブローカーと同じ ID を持つ一時ノードが登録されます。 Kafka のヘルスチェックはこのノードに依存します。これらのコンポーネントは、ブローカーがクラスターに参加したりクラスターから離脱したりするときに通知されます。

同じ ID を持つ別のブローカーを起動しようとすると、エラーが発生します。新しいブローカーは登録を試みますが、ZooKeeper に同じ ID を持つブローカーが既に存在するため、登録できません。
ブローカーがダウンしたり、パーティション分割されたり、ガベージコレクションの長時間の一時停止が発生すると、ブローカーは ZooKeeper から切断され、起動時にブローカーによって作成された一時ノードは ZooKeeper から削除されます。ブローカーリストをリッスンしている Kafka コンポーネントには、ブローカーが削除されたことが通知されます。
ブローカーをシャットダウンすると、対応するノードも消えますが、その ID はトピックのレプリカリストなどの他のデータ構造に引き続き存在します。レプリカリストのレプリケーションについては後ほど説明します。ブローカーを正常にシャットダウンした後、同じ ID で新しいブローカーを起動すると、そのブローカーは古いブローカーと同じパーティションとトピックを持つクラスターにすぐに参加します。

ブローカーコントローラーの役割

以前、Kafka Rebalance について説明した際に、グループ間の関係を調整する役割を担うグループコーディネーターについて説明しました。ブローカー間にはコントローラーコンポーネント (Controller) もあり、これが Kafka のコアコンポーネントとなります。その主な機能は、ZooKeeper の助けを借りて Kafka クラスター全体を管理および調整することです。クラスター内の各ブローカーはコントローラーと呼ぶことができますが、Kafka クラスターが起動すると、コントローラーになるブローカーは 1 つだけになります。 Kafka クラスターは ZooKeeper クラスターに依存しているため、ここで znode ノードの問題について簡単に説明します。

ZooKeeper データはノードに保存されます。各ノードは znode とも呼ばれます。 znode はツリー状のファイル構造であり、Linux オペレーティングシステムのファイルパスに非常に似ています。 ZooKeeper のルートノードは / です。

Znode は、データの永続化方法に応じて、一時ノードと永続ノードに分けられます。永続ノードは ZooKeeper ステータスの変更によって消えることはありませんが、一時ノードは ZooKeeper を再起動すると自動的に消えます。

znode ノードには Watcher メカニズムがあります。データが変更されると、ZooKeeper は Watcher イベントを生成し、クライアントに送信します。 Watcher 監視メカニズムは、Zookeeper の非常に重要な機能です。監視ノードデータの変更、ノードの削除、子ノードのステータスの変更、その他のイベントなど、監視イベントを Zookeeper に基づいて作成されたノードにバインドできます。このイベントメカニズムを通じて、分散ロック、クラスター管理、および ZooKeeper に基づくその他の機能を実装できます。

コントローラーの選出

Kafka の現在のコントローラー選出ルールは、Kafka クラスターで最初に起動されたブローカーが ZooKeeper に一時ノード/コントローラーを作成してコントローラーになることです。他のブローカーも起動時にこのノードを作成しようとしますが、このノードはすでに存在するため、後で /controller ノードを作成しようとすると、ノードがすでに存在するという例外が発生します。その後、他のブローカーはこのコントローラーに ZooKeeper ウォッチオブジェクトを登録します。 /controller ノードが変更されると、他のブローカーはノード変更通知を受信します。これにより、コントローラーが 1 つだけ存在するようになります。そうすると、単一のノードのみの問題、つまり単一ポイント問題が発生するはずです。

コントローラーがシャットダウンされるか、ZooKeeper との接続が失われると、ZooKeeper 上の一時ノードは消えます。クラスター内の他のノードが、コントローラがオフラインであるというメッセージを監視オブジェクトから受信すると、他のブローカーノードが新しいコントローラになろうとします。他のノードの作成ルールは、最初のノードの作成原則と同じです。 ZooKeeper でコントローラーノードを正常に作成した最初のブローカーが新しいコントローラーになります。その後、他のノードはノードがすでに存在するという例外を受け取り、監視のために新しいコントローラーノード上にウォッチオブジェクトを再度作成します。

コントローラーの役割

それで、ここまで話してきましたが、コントロールとは何でしょうか?コントローラーの機能は何ですか?あるいは、コントローラーのそのようなコンポーネントは何を行うために設計されているのでしょうか?心配しないでください。次にそれについてお話します。

Kafka は、ステートマシンをシミュレートするマルチスレッドコントローラーとして設計されています。以下のことが可能です:

コントローラは、部門（クラスタ）内の部門マネージャ（ブローカーコントローラ）に相当し、部門内の部門メンバー（ブローカー）を管理するために使用されます。
コントローラはすべてのブローカーのモニターであり、ブローカーのオンラインとオフラインを監視するために使用されます。
ブローカーがクラッシュした後、コントローラは新しいパーティションリーダーを選出できる。
コントローラはブローカーの新しく選択されたリーダーにメッセージを送信できる。

さらに細かく分けると、以下の5つのポイントに分けられます。

トピック管理: Kafka コントローラーは、Kafka トピックへのパーティションの作成、削除、追加に役立ちます。つまり、パーティションに対して最高の権限を持ちます。

つまり、kafka-topics スクリプトを実行すると、バックグラウンド作業のほとんどはコントローラーによって実行されます。

パーティションの再割り当て: パーティションの再割り当ては主に、kafka-reassign-partitions スクリプトによって提供される既存のトピックパーティションのきめ細かい割り当て機能を指します。この機能の部分もコントローラーによって実装されます。
優先リーダー選出: 優先リーダー選出は主に、一部のブローカーの過負荷を回避するために Kafka によって提供されるリーダー置換ソリューションです。
クラスタメンバー管理: 主に新しいブローカーの追加、ブローカーの終了、ブローカーのダウンタイムを管理します。
データサービス: コントローラーの最後の主要な作業カテゴリは、他のブローカーにデータサービスを提供することです。コントローラーは最も完全なクラスターメタデータ情報を保存します。他のすべてのブローカーは、コントローラからメタデータ更新要求を定期的に受信し、メモリ内のキャッシュデータを更新します。これらのデータについては以下で説明します。

コントローラーは、ブローカーがクラスターを離れたことを検出すると (関連する ZooKeeper パスを監視することによって)、このブローカーによって管理されるパーティションには新しいリーダーが必要であるというメッセージを受信します。コントローラーは各パーティションを順番に走査して、誰が新しいリーダーになれるかを判断し、新しいリーダーまたは既存のフォロワーを含むすべてのパーティションにメッセージを送信します。リクエストメッセージには、新しいリーダーとフォロワーが誰であるかに関する情報が含まれます。その後、新しいリーダーはプロデューサーとコンシューマーからのリクエストの処理を開始し、フォロワーは新しいリーダーからのレプリケーションに使用されます。

これはアウトソーシング会社の部門と非常によく似ています。この部署は出張を専門に扱っています。みんなそれぞれ違う場所で働いているんですが、中央本部に部長がいて、その部長が突然辞めてしまったんです。同社は外部から人材を採用する予定はなく、部門内から有能な人材をリーダーに選抜することにした。リーダーになった人はチームメンバーにメッセージを送る必要があります。このメッセージは任命メッセージであり、彼が誰を管理するのかを明確にします。誰もがそれを知って、それぞれの部署で働いています。

コントローラは、クラスターに参加するブローカーを検出すると、ブローカー ID を使用して、新しく参加したブローカーに既存のパーティションのレプリカが含まれているかどうかを確認します。コントローラーがある場合、メッセージは新しく追加されたブローカーと既存のブローカーに送信されます。

次に、パーティションレプリケーションに関する上記の内容について説明します。

ブローカーコントローラデータストレージ

前述のように、ブローカーコントローラーは大量の Kafka クラスターデータを保存するためのデータサービスを提供します。下記の通り

上記で保存された情報は、次の 3 つのカテゴリに分類できます。

ブローカー内のすべてのパーティション、ブローカーのすべてのパーティションレプリカ、現在実行中のブローカー、シャットダウン中のブローカーなど、ブローカーに関するすべての情報。
リーダーレプリカは誰か、どのレプリカが ISR セット内にあるかなど、特定のパーティション情報を含むすべてのトピック情報。
操作および保守タスクに関係するすべてのパーティション。現在優先リーダーの選出とパーティションの再割り当てが行われているパーティションのリストが含まれます。

Kafka は ZooKeeper と切り離せないため、このデータ情報のコピーも ZooKeeper に保存されます。コントローラーが初期化されるたびに、ZooKeeper から対応するメタデータが読み取られ、独自のキャッシュに入力されます。

ブローカーコントローラのフェイルオーバー

前述したように、ZooKeeper の /brokers/ids の下にノードを作成する最初のブローカーがブローカーコントローラーとして使用されます。つまり、ブローカーコントローラーは 1 つしかないため、単一障害点の問題が必然的に発生します。 Kafka は、この状況を考慮して、フェイルオーバーとも呼ばれるフェイルオーバー機能を提供します。以下のように表示されます。

最初は、broker1 が最初に正常に登録され、コントローラーになります。その後、ネットワークジッターまたはその他の理由により、broker1 はオフラインになります。 ZooKeeper は、Watch メカニズムを通じて broker1 のオフラインを検出します。その後、生き残ったブローカーはすべてコントローラーになるために競争を始めます。この時点で、broker3 が最初に正常に登録されます。このとき、ZooKeeper に保存されるコントローラー情報は broker1 -> broker3 になります。その後、broker3 は ZooKeeper からメタデータ情報を読み取り、それを独自のキャッシュに初期化します。

注意: ZooKeeper に保存されるのはキャッシュ情報ではありませんが、ブローカーに保存されるのはキャッシュ情報です。

ブローカーコントローラの問題

Kafka 0.11 より前では、コントローラーの設計は非常に面倒でした。上で述べたように、Kafka コントローラーは、ステートマシンをシミュレートするマルチスレッドコントローラーとして設計されています。このデザインには実はいくつか問題があります。

コントローラーの状態の変更は異なるリスナーによって同時に実行されるため、エラーが発生しやすくデバッグが困難な複雑な同期が必要になります。
状態の伝播は同期されておらず、ブローカーが不確実な時間に複数の状態になる可能性があり、これにより不必要な追加データ損失が発生する可能性があります。
コントローラーはトピック削除用の追加の I/O スレッドも作成するため、パフォーマンスが低下します。
コントローラーのマルチスレッド設計では、共有データにもアクセスします。ご存知のとおり、共有データへのマルチスレッドアクセスは、スレッド同期の最も厄介な部分です。データのセキュリティを保護するために、コントローラーはコード内で多数の ReentrantLock 同期メカニズムを使用する必要があります。これにより、コントローラー全体の処理速度がさらに低下します。

ブローカーコントローラの内部設計原則

Kafka 0.11 以降、Kafka コントローラーは新しい設計を採用し、マルチスレッドソリューションをシングルスレッドとイベントキューのソリューションに変更しました。次の図に示すように:

主な変更点は次のとおりです。

最初の改善点は、イベントエグゼキュータスレッドの追加です。図からわかるように、イベントキューとコントローラーコンテキストの両方が、処理のためにイベント実行スレッドに引き渡されます。最初に実行されたすべての操作は独立したイベントにモデル化され、このスレッドによる消費のために専用のイベントキューに送信されます。

2 番目の改善点は、これまでのすべての同期 ZooKeeper 操作を非同期操作に変更することです。 ZooKeeper API には、同期と非同期の 2 つの読み取りおよび書き込み方法が用意されています。以前は、コントローラーは ZooKeeper を同期的に操作していました。今回、同期モードから非同期モードに変更されました。テストによれば、効率は10倍向上した。

3 番目の改善点は、優先度に基づいてリクエストを処理することです。以前の設計では、ブローカーはコントローラーから送信されたすべてのリクエストを公平に処理していました。それはどういう意味ですか？公平であることは良いことではないでしょうか?場合によっては、そうです。たとえば、ブローカーが生成要求を処理するためにキューに入っている場合、コントローラーは StopReplica 要求を送信します。あなたは何をしますか？生産リクエストの処理を続行しますか?この生産リクエストはまだ役に立ちますか?この時点で、最も合理的な処理順序は、StopReplica 要求に高い優先度を与えて、事前に処理できるようにすることです。

コピーメカニズム

レプリケーションは Kafka アーキテクチャのコア機能です。 Kafka のドキュメントでは、Kafka は分散型、パーティション化可能、複製可能なコミットログサービスであると説明されています。レプリケーションが非常に重要である理由は、メッセージの永続的な保存が非常に重要であり、これによりプライマリノードがダウンした後でも Kafka の高可用性が確保されるためです。レプリケーションメカニズムは、バックアップメカニズム (レプリケーション) とも呼ばれ、通常は、ネットワークと対話する複数のマシンに同じデータのバックアップ/コピーを保存する分散システムを指します。

Kafka はトピックを使用してデータを整理します。各トピックは複数のパーティションに分かれています。パーティションは 1 つ以上のブローカーにデプロイされます。各パーティションには複数のコピーがあるため、コピーもブローカーに保存されます。各ブローカーは数千のコピーを保管する場合があります。次の図はレプリカレプリケーションの概略図です。

上の図に示すように、わかりやすくするためにブローカーは 2 つだけ描きました。各ブローカーはトピックのメッセージを保存します。 broker1 のパーティション 0 はリーダーであり、パーティションのレプリケーションを担当します。ブローカー 1 のパーティション 0 のコピーをブローカー 2 のトピック A のパーティション 0 にコピーします。同様に、トピック A のパーティション 1 についても同様です。

レプリカには、リーダーレプリカとフォロワーレプリカの 2 種類があります。

リーダーレプリカ

Kafka はパーティションを作成するときにレプリカを選択します。選出されたレプリカはリーダーレプリカです。

フォロワーインスタンス

リーダーレプリカ以外のレプリカは、総称してフォロワーレプリカと呼ばれます。フォロワーは外部サービスを提供しません。リーダーレプリカの動作は次のとおりです。

この写真について注目すべき点がいくつかあります。

Kafka では、Follower レプリカは外部サービスを提供しません。これは、フォロワーレプリカがコンシューマーとプロデューサーからの要求に応答できないことを意味します。すべてのリクエストはリーダーレプリカによって処理されます。つまり、すべてのリクエストは、リーダーのコピーが配置されているブローカーに送信する必要があります。フォロワーコピーは、リーダーとの同期を実現するために、データを非同期的にプルし、それを独自のコミットログに書き込むためにのみ使用されます。
リーダーレプリカが配置されているブローカーがダウンすると、Kafka は ZooKeeper が提供する監視機能を利用してそれをリアルタイムで検出し、フォロワーレプリカから 1 つをリーダーとして選択するための新しい選挙ラウンドを開始できます。クラッシュしたブローカーが再起動されると、パーティションのレプリカはフォロワーとして再参加します。

リーダーのもう一つの仕事は、どのフォロワーのステータスが自分のステータスと一致しているかを判断することです。リーダーの状態との一貫性を確保するために、フォロワーは新しいメッセージが到着する前にリーダーからのメッセージをコピーしようとします。リーダーに追いつくために、フォロワーはデータを取得するリクエストをリーダーに送信します。これは、コンシューマーがメッセージを読み取るために送信するメッセージと同じです。

フォロワーがリーダーにメッセージを送信するプロセスは、最初にメッセージ 1 を要求し、次にメッセージ 1 を受信し、要求 1 の時間が来たら要求 2 を送信します。フォロワーは、リーダーからフォロワーへのメッセージを受信するまで、メッセージの送信を続行しません。プロセスは次のとおりです。

フォロワーレプリカは、応答メッセージを受信するまでメッセージの送信を継続しないことが重要です。各フォロワーが要求した最新のオフセットを確認することで、リーダーは各フォロワーのレプリケーションの進行状況を把握できます。フォロワーが 10 秒以内にメッセージをリクエストしなかった場合、またはフォロワーがリクエストを送信したが 10 秒以内にメッセージを受信しなかった場合は、同期されていないとみなされます。レプリカがリーダーと同期されていない場合、リーダーがオフラインになると、このレプリカのメッセージは完了していないため、このレプリカはリーダーとは呼ばれなくなります。

逆に、フォロワーによって同期されたメッセージがリーダーレプリカのメッセージと一致する場合、このフォロワーレプリカは同期レプリカと呼ばれます。つまり、リーダーがオフラインになった場合、同期されたレプリカだけがリーダーになることができます。

これまでコピーメカニズムについて詳しく説明してきましたが、コピーメカニズムの利点は何でしょうか?

書き込まれたメッセージはすぐに確認できます。つまり、プロデューサー API を使用してパーティションにメッセージを正常に書き込んだ後、すぐにコンシューマーを使用して、書き込まれたメッセージを読み取ることができます。
メッセージの冪等性を実現するとはどういう意味ですか?つまり、プロデューサーによって生成されたメッセージについては、コンシューマーがそれらを消費するときに、そのメッセージの存在が毎回確認され、メッセージが存在しない状況は発生しません。

同期レプリケーションと非同期レプリケーション

レプリケーションのメカニズムについて学んでいたとき、疑問が浮かびました。リーダーレプリカとフォロワーレプリカは同期レプリケーション方式である送信待機メカニズムを使用しているのに、フォロワーレプリカがリーダーレプリカと同期するときに非同期操作であると言われるのはなぜですか?

こんな感じだと思います。リーダーコピーを同期した後、フォロワーコピーはメッセージをローカルログに保存します。このとき、フォロワーはリーダーのコピーに応答メッセージを送信し、リーダーに正常に保存されたことを伝えます。同期レプリケーションのリーダーは、すべてのフォロワーコピーが正常に書き込まれるまで待機し、書き込みが成功したことを示すメッセージをプロデューサーに返します。非同期レプリケーションでは、リーダーレプリカはフォロワーレプリカが正常に書き込まれたかどうかを気にする必要はありません。リーダーレプリカがメッセージをローカルログに保存する限り、書き込みが成功したというメッセージがプロデューサーに返されます。同期レプリケーションと非同期レプリケーションのプロセスは次のとおりです。

同期レプリケーション

プロデューサーはリーダーを識別するためにZooKeeperに通知する
プロデューサーがリーダーにメッセージを書く
メッセージを受け取ったリーダーはそれをローカルログに書き込みます。
フォロワーはリーダーからのメッセージを受け取る
フォロワーはローカルにログを書き込む
フォロワーはリーダーに書き込み成功メッセージを送信する
リーダーはすべてのフォロワーからメッセージを受信します。
リーダーは書き込み成功メッセージをプロデューサーに送信する

非同期レプリケーション

同期レプリケーションとの違いは、リーダーがローカルログに書き込んだ後、すべてのフォロワーがレプリケーションを完了するのを待たずに、書き込み成功メッセージをクライアントに直接送信することです。

情報サービス

Kafka は、ISR と呼ばれる同期レプリカのセットを動的に維持します。 ISR も非常に重要な概念です。前述したように、フォロワーレプリカはサービスを提供せず、リーダーレプリカからデータを定期的かつ非同期的に取得するだけです。プル操作はコピー操作と同じです。 ctrl-c + ctrl-v は皆さんもよくご存知だと思います。これは、ISR セット内のレプリカメッセージの数がリーダーレプリカメッセージの数と同じになることを意味しますか?必ずしもそうではありません。判断の基準は、ブローカーのパラメータ replica.lag.time.max.ms の値です。このパラメータは、フォロワーレプリカがリーダーレプリカより遅れることができる最長時間間隔を意味します。

replica.lag.time.max.ms パラメータのデフォルト時間は 10 秒です。フォロワーレプリカがリーダーレプリカより 10 秒以内遅れている場合、Kafka はリーダーとフォロワーが同期されていると見なします。この時点でも、フォロワーコピーに保存されるメッセージはリーダーコピーに保存されるメッセージよりも小さくなります。フォロワーレプリカがリーダーレプリカより 10 秒以上遅れている場合、フォロワーレプリカは ISR から削除されます。レプリカが後でゆっくりとリーダーに追いついた場合は、ISR に再度追加できます。これは、ISR が静的なセットではなく、動的に調整されるセットであることも示しています。

不潔なリーダー選挙

ISR は動的に調整できるため、ISR セットが空になる状況が必然的に発生します。リーダーコピーは ISR セットに表示される必要があるため、ISR セットが空であるということは、リーダーコピーもダウンしていることを意味するため、この時点で Kafka は新しいリーダーを再選出する必要があります。では、新しいリーダーをどうやって選出するのでしょうか?今、あなたは考え方を変える必要があります。上記で、ISR セットはリーダーと同期されたコピーである必要があると述べました。したがって、ISR セットになくなったコピーはリーダーと同期されていないコピーである必要があります。つまり、ISR リストになくなったフォロワーコピーは一部のメッセージを失います。ブローカー側のパラメータ unclean.leader.election.enable を有効にすると、これらの非同期レプリカの中から次のリーダーが選出されます。このタイプの選挙は、アンクリーンリーダー選挙とも呼ばれます。

分散プロジェクトに携わったことがあるなら、CAP 理論を知っているはずです。このクリーンでないリーダー選出は、実際には Kafka の高可用性を確保するためにデータの一貫性を犠牲にします。

実際のビジネスシナリオに基づいて、アンクリーンリーダー選出を有効にするかどうかを決定できます。データの一貫性は可用性よりもはるかに重要であるため、通常、このパラメータを有効にすることは推奨されません。

Kafka リクエスト処理フロー

ブローカーの作業の大部分は、クライアント、パーティションレプリカ、およびコントローラーからパーティションリーダーへの要求を処理することです。このタイプのリクエストは通常、リクエスト/レスポンスタイプです。最初に公開されるリクエスト/レスポンスメソッドは HTTP リクエストであるはずです。実際、HTTP リクエストは同期または非同期にすることができます。通常の HTTP リクエストは通常、同期されます。同期方式の最大の特徴は、リクエストの送信→サーバーの処理待ち→処理後の戻りまでの期間中、クライアントブラウザは何もできないことです。非同期方式の最大の特徴は、リクエストがイベントによってトリガーされ、サーバーによって処理される（この時点ではブラウザはまだ他の処理を実行できます）→ 処理されるという点です。

また、同期リクエストは順番に処理されるのに対し、非同期リクエストの実行方法は不確定であると言えます。非同期では複数の実行スレッドの作成が必要であり、各スレッドの実行順序が異なるためです。

ここで注意すべき点は、例として HTTP リクエストのみを使用しているのに対し、Kafka は通信にソケットに基づく TCP を使用していることです。

では、これら 2 つの方法の欠点は何でしょうか?

賢い人ならすぐに思いつくはずだと信じています。同期方式の最大の欠点は、スループットが非常に低く、リソースの使用率が極めて低いことです。リクエストは順番にしか処理できないため、各リクエストは前のリクエストが処理されるまで待機してから処理される必要があります。このアプローチは、リクエストが非常にまれにしか送信されないシステムにのみ適しています。

非同期アプローチの欠点は、リクエストごとにスレッドを作成するのに非常にコストがかかり、場合によってはサービス全体に負担がかかってしまう可能性があることです。

レスポンシブモデル

長々と話しましたが、Kafka は同期ですか、それとも非同期ですか?どちらでもない。 Kafka はレスポンシブ (Reactor) モデルを採用しています。では、レスポンシブモデルとは何でしょうか?簡単に言えば、Reactor モードはイベント駆動型アーキテクチャの実装であり、次の図に示すように、複数のクライアントが同時にサーバーに要求を送信する処理シナリオに特に適しています。

Kafka のブローカー側には、プロセッサに似た SocketServer コンポーネントがあります。 SocketServer は TCP ソケット接続に基づいており、クライアント要求を受け入れるために使用されます。すべてのリクエストメッセージにはメッセージヘッダーが含まれており、次の情報が含まれています。

リクエストタイプ（APIキーとも呼ばれます）
リクエストバージョン (ブローカーはクライアントリクエストの異なるバージョンを処理し、クライアントバージョンに基づいて異なる応答を行うことができます)
相関ID --- リクエストメッセージを識別するために使用される一意の番号。応答メッセージとエラーログにも表示されます（問題の診断に使用されます）。
クライアントID --- リクエストを送信するクライアントを識別するために使用されます

ブローカーは、リッスンする各ポートで Acceptor スレッドを実行します。このスレッドは接続を作成し、それをプロセッサ (ネットワークスレッドプール) に渡します。プロセッサの数は、num.network.threads を使用して設定できます。デフォルト値は 3 です。つまり、各ブローカーはクライアントから送信されたリクエストの処理を開始するときに 3 つのスレッドを作成します。

アクセプタースレッドはポーリングを使用して、スタックされた要求をネットワークスレッドプールに公平に送信します。したがって、実際の使用では、これらのスレッドは通常、保留中の要求キューに割り当てられる確率が同じであり、その後、応答キューから応答メッセージを取得してクライアントに送信します。プロセッサネットワークスレッドプールでの要求応答処理は比較的複雑です。以下は、ネットワークスレッドプールでの処理フローチャートです。

プロセッサネットワークスレッドプールは、クライアントや他のブローカーからメッセージを受信すると、そのメッセージを要求キューに格納します。これは共有リクエストキューであることに注意してください。ネットワークスレッドプールはマルチスレッドメカニズムであるため、要求キュー内のメッセージは複数のスレッドによって共有されます。次に、IO スレッドプールはメッセージを処理し、メッセージの種類に基づいて何を実行するかを決定します。たとえば、PRODUCE 要求の場合、メッセージはログに書き込まれます。 FETCH 要求の場合、メッセージはディスクまたはページキャッシュから読み取られます。つまり、IO スレッドプールは、実際に判断を下し、リクエストを処理するコンポーネントです。 IO スレッドプールは処理を完了すると、それを応答キューに配置するか、Purgatory に配置するかを決定します。煉獄とは何かについては以下で説明します。それでは、応答キューについて説明します。応答キューは各スレッドに固有です。レスポンシブモデルでは、リクエストがどこに送信されるかは考慮されないため、レスポンスの戻りは各スレッドに任され、共有する必要はありません。

注: IO スレッドプールは、ブローカー側のパラメータ num.io.threads を通じて構成できます。デフォルトのスレッド数は 8 です。つまり、各ブローカーの起動後に 8 つの IO 処理スレッドが自動的に作成されます。

リクエストタイプ

一般的なリクエストの種類は次のとおりです。

生産依頼

簡単に言えば、構成が異なれば、成功した書き込みの定義も異なります。 acks = 1 の場合、リーダーがメッセージを受信する限り、書き込みは成功したことを意味します。 acks = 0 の場合、リーダーがメッセージを送信すれば書き込みが成功したことを意味し、戻り値の影響を考慮する必要はありません。 acks = all の場合、リーダーは書き込みが成功したことを示す前にすべてのレプリカからメッセージを受信する必要があることを意味します。

メッセージがパーティションのリーダーに書き込まれた後、acks 構成の値が all の場合、リーダーレプリカがフォロワーレプリカがメッセージをコピーしたことを検出するまで、これらの要求は Purgatory バッファーに保存され、応答がクライアントに送信されます。

リクエストを取得

ブローカーがリクエストを取得する方法は、プロダクションリクエストを処理する方法と似ています。クライアントは、トピックパーティション内の特定のオフセットにあるメッセージの要求をブローカーに送信します。オフセットが存在する場合、Kafka はゼロコピー技術を使用してメッセージをクライアントに送信します。 Kafka は、バッファを経由せずにファイルからネットワークチャネルにメッセージを直接送信するため、パフォーマンスが向上します。

クライアントは要求するデータの上限と下限を設定できます。上限とは、十分なメッセージを受信するためにクライアントによって割り当てられたメモリ領域を指します。この制限はより重要です。上限が大きすぎると、クライアントのメモリが直接使い果たされる可能性が高くなります。下限値は、送信する前に十分なデータパケットを蓄積することを意味します。これは、プロジェクトマネージャーがプログラマーに 10 個のバグを割り当てるのと同じです。プログラマーは、バグを修正するたびにプロジェクトマネージャーに報告します。修正が完了することもあれば、そうでない場合もあります。これにより、通信コストと時間コストが増加します。したがって、下限は、プログラマーが10個のバグを修正した後に私に報告する必要があることです!!!次の図に示すように:

図で見ることができるように、メッセージを引く間にメッセージの蓄積を待つプロセス--->メッセージの送信があります。このメッセージの蓄積はタイムアウトと考えることができます。ただし、タイムアウトが発生した場合に例外が発生し、メッセージの蓄積タイムアウト後に領収書が応答されます。遅延時間は、Replica.lag.time.max.msで構成できます。これは、メッセージを複製するときにレプリカが許可される最大遅延時間を指定します。

メタデータリクエスト

生産要求と応答要求の両方をリーダーレプリカに送信する必要があります。ブローカーが特定のパーティションのリクエストを受け取り、リクエストのリーダーが別のブローカーにある場合、リクエストを送信するクライアントは、非党のリーダーからエラー応答を受け取ります。パーティションのリクエストがリーダーを含まないブローカーに送信されると、同じエラーが発生します。 Kafkaクライアントは、正しいブローカーにリクエストと応答を送信する必要があります。これはナンセンスではないでしょうか?どこに送るかを知るにはどうすればよいですか？

実際、クライアントは、クライアントが関心を持っているトピックのリストを含むメタデータリクエストを使用し、サーバーの応答メッセージは、トピックのパーティション、リーダーのレプリカ、フォロワーレプリカを示します。すべてのブローカーがこの情報をキャッシュするため、メタデータリクエストは任意のブローカーに送信できます。

一般に、クライアントはこの情報をキャッシュし、ターゲットブローカーに生産リクエストと応答リクエストを直接送信します。これらのキャッシュは、Metadata.max.age.msパラメーターを使用して、メタデータが変更されたかどうかを構成するために、定期的にリフレッシュする必要があります。たとえば、新しいブローカーが参加すると、リバランスがトリガーされ、いくつかのレプリカが新しいブローカーに移動されます。この時点で、クライアントがリーダーではないというエラーを受け取った場合、クライアントはリクエストを送信する前にメタデータキャッシュを再表示します。

Kafkaのリバランスプロセス

記事の消費者について本当に説明したとき、この記事を読んでカフカを始めただけで、消費者グループとリバランスの関係について簡単に話しました。実際、グループ内のすべての消費者インスタンスがどのトピックパーティションを消費するかについて同意するように要約することができます。

消費者グループにグループコーディネーターが存在するに違いないことを知っており、コーディネーターの助けを借りてリバランスプロセスが完了します。

ここでは、リバランスの条件を述べる必要があります。

消費者が購読するトピックは変更されます
消費者数の変化
パーティションの数が変更されました
まだ作成されていないトピックを購読すると、トピックが作成されたときにリバランスが発生します。購読するトピックが削除されている場合、リバランスも発生します
消費者は、グループコーディネーターによって死んでいると見なされます。これは、消費者がクラッシュするか、長い間走っている状態にあるために発生する可能性があります。これは、消費者が合理的な構成時間内にグループコーディネーターにハートビートを送信していないことを意味します。これはまた、リバランスを引き起こします。

リバランスを理解する前に、これらの2つの役割を知る必要があります

グループコーディネーター：グループコーディネーターは、消費者グループのすべての消費者からハートビートメッセージを受信するブローカーです。初期のバージョンでは、メタデータ情報はZookeeperに保存されていましたが、現在メタデータ情報はブローカーに保存されています。各消費者グループは、グループのグループコーディネーターと同期する必要があります。すべての意思決定はアプリケーションノードで行われますが、グループコーディネーターはJoingroupリクエストを満たし、割り当てやオフセットなどの消費者グループに関するメタデータ情報を提供できます。グループコーディネーターには、すべての消費者の鼓動を知る権利もあります。また、リーダーである消費者グループにも役割があります。リーダーレプリカやKafkaコントローラーと区別することに注意してください。リーダーはグループの意思決定の役割であるため、リーダーがオフラインになった場合、グループコーディネーターはすべての消費者をグループから追い出す権利を持っています。したがって、消費者グループの重要な行動は、リーダーを選出し、コーディネーターとの割り当てとパーティションに関するメタデータ情報を読み書きすることです。

消費者リーダー：各消費者グループにはリーダーがいます。消費者がハートビートの送信を停止した場合、コーディネーターはリバランスをトリガーします。

リバランスを理解する前に、状態マシンが何であるかを知る必要があります

Kafkaは、コーディネーターがリバランスプロセス全体を完了するのを支援するために、Consumer Group State Machineを設計しました。消費者州のマシンには、空、死亡、準備、re延の完了、安定性の5つの主要な状態があります。

これらの状態の意味を理解した後、消費者状態の回転を表すためにいくつかのパスを使用しましょう。

1.消費者グループは、最初は空の状態にあります。リバランスがオンになると、新しい消費者が参加するのを待っている準備状態の状態に配置されます。新しい消費者が参加すると、消費者グループは、割り当てを待っている完全なバランス状態になります。新しい消費者がグループに参加したり、休暇をとったりする限り、リバランスが引き起こされます。消費者の状態は、再生バランス状態にあります。割り当てメカニズムが指定され、割り当てが完了するのを待ち、そのフローチャートはこのようなものです。

2。上記の数値に基づいて、すべての消費者グループが安定した状態に到達すると、新しい消費者が参加/休暇/ハートビートの有効期限が切れると、リバランスがトリガーされ、消費者グループの状態が再び準備状態になります。その後、そのフローチャートはこのように見えます。

3.上記の数字に基づいて、消費者グループが準備状態にある後、残念ながら、誰もプレイせず、すべての消費者が去りました。現時点では、消費者消費の変位データが保持される場合があります。変位データの有効期限が切れるか、更新されると、消費者グループは死んだ状態になります。そのフローチャートは次のとおりです。

4。上記の図に基づいて、消費者のリバランスを分析しました。変位テーマのパーティションリーダーが、reprabalanceを準備し、再障害または安定した状態で変化する場合、グループは直接死んだ状態になり、そのすべてのパスは次のとおりです。

ここに注意すべき2つのポイント：

一般に、XXX何百万秒で必要なXXの有効期限が切れるオフセットの出現は、Kafkaがグループの変位データを削除する可能性が高いことを示しています。

空の状態のグループのみが、有効期限変位削除操作を実行します。

リバランスプロセス

上記では、消費者グループのステータスの変換プロセスについて学びましたが、これでリバランスのプロセスの導入を実際に開始します。リバランスプロセスは、消費者側とコーディネーター側の2つの側面から見ることができます。まず、消費者側を見てみましょう。

消費者はバランスを評価します

消費者のバランスに重点を置くという観点から2つのステップがあります。消費者はグループに参加し、リーダーが計画を割り当てるのを待ちます。これらの2つのステップの後の対応する要求は、JoingRoupとSyncgroupです。

新しい消費者がグループに参加すると、消費者はコーディネーターにJoingroupリクエストを送信します。このリクエストでは、各消費者メンバーが消費するトピックを提出する必要があります。上記のグループコーディネーターで述べたように、この目的は、コーディネーターが消費者グループのリーダーを選択するのに十分なメタデータ情報を収集できるようにすることです。通常、Joingroupリクエストを送信する最初の消費者は、リーダーと自動的に呼ばれます。リーダーのタスクは、すべてのメンバーからサブスクリプション情報を収集し、この情報に基づいて、特定のパーティション消費配分計画を策定することです。図に示すように:

すべての消費者がメタデータ情報に参加してリーダーに提出した後、リーダーは割り当て計画を立て、グループの消費戦略の発行を担当するコーディネーターに同期リクエストを送信します。次の図は、同期リクエストプロセスを説明しています。

すべてのメンバーが割り当て計画を正常に受け取ると、消費者グループは安定した状態に入ります。つまり、通常の消費作業を開始します。

コーディネーターの観点からバランスに焦点を当てます

コーディネーターの観点から見ると、主にバランスに焦点を当てるための次のトリガー条件があります。

新しいメンバーがグループに参加します

グループメンバーは自発的に去ります

グループメンバーがクラッシュして出発します

グループメンバーは変位を提出します

グループに参加する新しいメンバーから始めて、それを別々に説明しましょう

新しいメンバーが参加します

私たちが議論しているシナリオは、割り当てを待つ安定したプロセスです。この時点で、新しいメンバーがグループに参加した場合、リバランスのプロセスはリバランスです。

この観点から、コーディネーターのプロセスは消費者のプロセスに似ていますが、今は消費者の観点から、そして今はリーダーの観点からです

グループメンバーは去ります

グループメンバーは、消費者グループを去ることは、消費者インスタンスを指し、exting（）メソッドを積極的にコーディネーターに通知したいことを積極的に通知します。ここでは、reversggroup（）に表示される新しいリクエストがあります。次の図に示すように:

グループメンバーがクラッシュします

グループメンバーのクラッシュとは、消費者インスタンスの深刻な障害、ダウンタイム、または一定期間応答しないことを指し、コーディネーターは消費者の鼓動を受け取ることができません。これはグループメンバーのクラッシュと見なされます。グループを離れると、クラッシュは受動的です。コーディネーターは通常、それを感知するためにしばらく待つ必要があります。この期間は、一般的に消費者パラメーターSession.Timeout.msによって制御されます。次の図に示すように:

リバランス中に変位を提出します

このプロセスを表すためにグラフを使用しなくなります。要約すると、消費者がJoingroupリクエストを送信した後、グループの消費者は指定された時間範囲内でそれぞれの変位を提出し、通常のJoingroup/Syncgroupリクエストをオンにして送信する必要があります。

記事のリファレンス：

「カフカの権威あるガイド」
https://blog.csdn.net/u013256816/article/details/80546337
https://learning.oreilly.com/library/view/kafka-the-definitive/9781491936153/ch05.html#kafka_internals
https://www.cnblogs.com/kevingrace/p/9021508.html
https://www.cnblogs.com/huxi2b/p/6980045.html
「Geek Time-Kafkaコアテクノロジーと実用的なテクニック」
https://cwiki.apache.org/confluence/display/kafka/kafka+controller+redesign
https://cwiki.apache.org/confluence/display/kafka/kafka+Controller+Internals
Kafkaパーティションとレプリカ、Kafkaの実行プロセス、およびメッセージの高い可用性
HTTPでの同期および非同期リクエスト
原子炉モードの詳細な説明
https://kafka.apache.org/documentation/
https://www.linkedin.com/pulse/partitions-rebalance-kafka-raghunanandan-gupta
https://cwiki.apache.org/confluence/display/kafka/kafka+detailed+consumer+coordinator+design

【編集者のおすすめ】