マイクロサービスアーキテクチャ - 最も包括的な知識ポイントを備えたメッセージキュー Kafka ダイアグラム

MQ (メッセージキュー) は、非同期 RPC として理解できるプロセス間通信の方法の 1 つです。通話結果に対する上流システムの態度は、多くの場合重要ですが、緊急ではありません。メッセージキューを使用すると、ビジネスの分離、トラフィックのピーク削減、柔軟な拡張などの利点が得られます。次にメッセージミドルウェアKafkaを紹介します。

[[277525]]

Kafka とは何ですか?

Kafka は分散メッセージングエンジンです。以下の特徴があります

メッセージストリームを公開およびサブスクライブする機能 (メッセージキューと同様)
メッセージストリームをフォールトトレラントかつ永続的な方法で保存する
マルチパーティションコンセプトにより並列処理能力が向上

Kafka アーキテクチャの概要

トピック

メッセージトピックとキュー。各メッセージにはトピックがあり、Kafka はメッセージをトピックごとに分類します。 Kafka では、トピックは物理的に 1 つ以上のパーティションに分割できます。各パーティションは物理的に「topicName_partitionIndex」という名前のフォルダーに対応します。このディレクトリには、このパーティションのすべてのメッセージ (.log) とインデックスファイル (.index) が含まれており、これにより Kafka のスループットを水平方向に拡張できます。

パーティション

各パーティションは連続した不変のメッセージキューであり、継続的に追加できます。パーティション内のメッセージには、オフセットと呼ばれる、パーティションごとに一意のシーケンス番号が割り当てられます。

メッセージを公開するときに、プロデューサーは各メッセージのキーを指定できます。メッセージがブローカーに送信されると、パーティション分割アルゴリズムに従って対応するパーティションに保存されます (1 つのパーティションに複数のメッセージが格納されます)。パーティション分割ルールが適切に設定されていれば、すべてのメッセージが異なるパーティションに均等に分散され、負荷分散が実現されます。

ブローカ

メッセージの保存に使用される Kafka サーバー。 Kafka クラスター内の各サーバーはブローカーであり、コンシューマーはブローカーからサブスクライブされたメッセージをプルします。

プロデューサー

Kafka にメッセージを送信すると、プロデューサーはトピックに応じてメッセージを配布します。プロデューサーは、メッセージをトピック上のパーティションに関連付ける責任も負います。最も簡単な方法は、パーティションのリストから順番に選択することです。何らかのアルゴリズムに従って重みに基づいてパーティションを選択することもできます。アルゴリズムは開発者が定義できます。

消費者

Consermer インスタンスは、メッセージのサブスクライブと消費を担当する独立したプロセスになることができます。コンシューマーは、consumerGroup を使用して自分自身を識別します。同じコンシューマーグループは複数のパーティションからのメッセージを同時に消費することができ、同じパーティションを複数のコンシューマーグループが同時に消費することもできますが、コンシューマーグループ内のパーティションは 1 つのコンシューマーのみが消費できます。

顧客グループ

コンシューマーグループ: 同じコンシューマーグループ内のコンシューマーの場合、Kafka は対応するトピック内の各メッセージを 1 つのコンシューマーにのみ送信します。

Kafka プロデューサーの設計原則

メッセージを送信するプロセス

1. メッセージをシリアル化し、パーティションを計算する

キーと値の構成に従ってメッセージをシリアル化し、パーティションを計算します。

ProducerRecord オブジェクトでパーティションが指定されている場合は、このパーティションが使用されます。それ以外の場合は、キーの係数とトピックのパーティション数を取得します。キーが存在しない場合は、ランダムにカウンターを生成し、このカウンターを使用してパーティション数の係数を取得します。このカウンターは使用されるたびに増加します。

2バッチに送信&送信スレッドを起動

トピックパーティションに応じて対応するバッチを取得します（期限）、メッセージをバッチに追加します。バッチがいっぱいの場合は、送信者スレッドを起動します。キューの操作はロックを使用して実行されるため、バッチ内のメッセージは順序付けられます。現在のメソッドの後続の Sender 操作は非同期操作です。

3.送信者は順番にブローカーにメッセージを送信します（tp repliaリーダー）

3.1 TP Relicaリーダーが所在するブローカーを特定する

Kafka の各ブローカーは、Kafka クラスターのメタデータ情報を保存します。メタデータ情報には、各トピックのすべてのパーティションの情報（リーダー、リーダーエポック、コントローラーエポック、isr、レプリカなど）が含まれます。Kafka クライアントは、任意のブローカーから必要なメタデータ情報を取得できます。送信スレッドは、メタデータ情報を通じて TP リーダーの brokerId を知ることができます。
プロデューサーはメタデータ情報を保存し、メタデータ更新戦略（metadata.max.age.msの定期的な更新、障害検出、強制更新：メタデータが無効であることを確認した後、metadata.requestUpdate()を呼び出して強制的に更新）に基づいてメタデータを更新します。

パブリッククラスPartitionInfo{ 
 
プライベート最終文字列トピック; 
 
プライベート最終intパーティション; 
 
プライベート最終ノードリーダー。 
 
プライベート最終Node[]レプリカ; 
 
プライベート最終Node[] inSyncReplicas; 
 
プライベート最終Node[] offlineReplicas; 
 
 }

3.2 冪等送信

プロデューサーの冪等性を実現するために、Kafka はプロデューサー ID (PID) とシーケンス番号を導入しました。各PIDごとにプロデューサーは各メッセージを送信する

メッセージのシーケンス番号がブローカーによって保持されているシーケンス番号と大きく異なる場合は、一部のデータがまだ書き込まれていない、つまり順序が正しくないことを意味します。このとき、ブローカーはメッセージを拒否し、プロデューサーは InvalidSequenceNumber をスローします。
メッセージのシーケンス番号がブローカーによって保持されるシーケンス番号以下の場合、メッセージが保存されていること、つまり重複メッセージであることを意味します。ブローカーはメッセージを直接破棄し、プロデューサーは DuplicateSequenceNumber をスローします。
送信者は送信が失敗した後に再試行し、すべてのメッセージをブローカーに送信できるようにします。

4. 送信者はブローカーから送信されたプロデュース応答を処理する

ブローカーが送信者のプロデュース要求を処理すると、送信者にプロデュース応答を送信し、プロデューサーは send() に設定したコールバック関数を実行します。この時点でプロデューサーの送信は完了です。

スループットとレイテンシ:

buffer.memory: バッファ設定を大きくするとスループットが向上しますが、バッチサイズを大きくするとレイテンシが増加します。これは linger_ms パラメータと共に使用できます。
linger_ms: バッチが大きすぎる場合、またはプロデューサーの qps が高くない場合は、バッチはゆっくりと追加されます。 linger_ms 時間後にバッチデータを強制的に送信することができます。
ack: 配達が成功したとみなされるには、プロデューサーはブローカーからいくつの返信を受け取る必要がありますか?
0 はプロデューサーがリーダーの確認を待つ必要がないことを意味します (スループットは最高、データの信頼性は最低)
1は、リーダーがローカルログへの書き込みを確認し、すぐに確認する必要があることを意味します。
-1/all は、すべての ISR が完了した後の確認を意味します (スループットは最低、データの信頼性は最高)

送信スレッドと長い接続

プロデューサーインスタンスが初期化されるたびに、センダーインスタンスが初期化され、ブローカーに永続的な接続が追加されます。

コードの観点: KafkaProducerが初期化されるたびに、空のクライアントが割り当てられます

パブリックKafkaProducer(最終マップ構成) {
 this(configs, null , null , null , null , null , Time .SYSTEM);
 }

端末上の TCP 接続の数を確認します。

 lsof -p ポート番号 -np | grep TCP、プロデューサーの数を適切に増やすことでスループットが向上する可能性がある

消費者デザインの原則

投票メッセージ

コンシューマーはフェッチスレッド（シングルスレッド）を通じてメッセージをプルします。
コンシューマーはハートビートスレッドを通じてブローカーにハートビートを送信します。タイムアウトを超えるとハングアップとみなされます
各コンシューマーグループには、それを管理するブローカーのコーディネーターがいます。コンシューマーの参加と退出、および消費されたメッセージの置き換えはすべてコーディネーターによって処理されます。

移転管理

コンシューマーのメッセージオフセットは、トピックパーティションの現在のグループの消費の進行状況を表します。コンシューマーがクラッシュして再起動した後も、このオフセットから引き続き消費できます。 Kafka0.8 より前では、オフセット情報は Zookeeper に保存されていました。 Zookeeper は高同時実行の読み取りと書き込みには適していないため、新しいバージョンの Kafka はオフセット情報をメッセージとして扱い、__consumers_offsets トピックが配置されているブローカーに送信します。 __consumers_offsets にはデフォルトで 50 個のパーティションがあります。メッセージのキーは groupId+topic_partition で、値は offset です。

Kafka グループのステータス

空: 初期状態。グループにはメンバーがいません。すべてのオフセットが期限切れになると、Dead になります。
再調整の準備: グループは再調整の準備を進めています
AwaitingSync: グループはグループリーダーからの割り当て計画を待機しています。
安定：安定した状態（グループが安定している）
停止中: グループにメンバーが存在せず、メタデータが削除されています。
知らせ

リバランス

何らかの理由でコンシューマーのパーティション消費が均一でなくなった場合、Kafka は自動的に再バランスを実行し、コンシューマーのパーティション消費を再び均等にします。

リバランスはいつ発生しますか?

グループに登録されているトピックの数が変わる
トピックパーティションの数を変更する
消費者会員の変更
消費者がグループに参加または離脱するとき
消費者がクラッシュしたと検出された場合

リバランスプロセス

例1: 消費者がクラッシュしたと検出されたことによる再バランス

たとえば、ハートビートスレッドがタイムアウト期間内にブローカーにハートビートを送信しない場合、コーディネーターはグループの再バランスをとる必要があると判断します。次に、他のコンシューマーがフェッチ要求を送信すると、コーディネーターは再バランス通知で応答します。コンシューマーメンバーがリクエストを受信すると、リーダーのみが割り当て戦略に従って割り当てを行い、それぞれの割り当て結果をコーディネーターに返します。このとき、コンシューマーリーダーのみが実際のデータを返し、その他は空のデータを返します。割り当て方法を受け取った後、消費者は割り当て戦略を各消費者に同期します。

例2: 消費者の参加による再バランス

参加プロトコルを使用して、消費者がグループに参加したいことを示す
同期プロトコルを使用して割り当てルールに従って割り当てる

（上の写真はインターネットから引用）

拡張: 上記の再バランス調整メカニズムの問題点

大規模なシステムでは、トピックは数百のコンシューマーインスタンスに対応する場合があります。これらの消費者が次々と空の消費者グループに参加すると、複数の再バランスが発生します。さらに、コンシューマーインスタンスの起動時間は制御不能であり、コーディネータによって決定された再バランスタイムアウト (つまり、max.poll.interval.ms) を超える可能性があり、これにより再バランスが再度トリガーされます。再バランスの前に多くの状態を永続化し、再バランス後に再初期化する必要があるため、各再バランスのコストは非常に高くなります。

新バージョンの改善

PreparingRebalance状態への移行を遅らせることで、リバランスの回数を減らす

新しいバージョンでは、group.initial.rebalance.delay.ms パラメータが追加されました。空のコンシューマーグループがメンバー参加リクエストを受信しても、すぐに PreparingRebalance 状態に切り替わって再バランスが開始されるわけではありません。時間が group.initial.rebalance.delay.ms を超えると、グループのステータスが PreparingRebalance (再バランス開始) に変更されます。実装メカニズムは、コーディネーターの最下層に新しいグループ状態 InitialReblance を追加することです。この時点で複数のコンシューマーが次々に起動されると仮定すると、グループの状態は最初に InitialRebalance に変換され、その後、group.initial.rebalance.delay.ms 時間後に PreparingRebalance (再バランスを開始) に変換されます。

ブローカー設計原則

ブローカーは Kafka クラスター内のノードです。プロデューサーによって送信されたメッセージとコンシューマーによる消費要求の処理を担当します。クラスタノードの管理など内容がたくさんあるので、まずは簡単に紹介し、後ほど特別記事でシェアしたいと思います。

ブローカーzk登録

ブローカーメッセージストレージ

Kafkaメッセージはバイナリ形式でコンパクトに保存されるため、多くのスペースを節約できます。
さらに、メッセージはヒープではなく ByteBuffer に保存されるため、ブローカープロセスがハングアップしてもデータが失われず、GC の問題が回避されます。
ゼロコピーとシーケンシャルアドレス指定により、メッセージの保存と読み取りが非常に高速になります。
ゼロコピーによるフェッチリクエストの高速化

ブローカーステータスデータ

ブローカー設計では、各マシンは同じ状態データを保存します。主に以下の内容が含まれます。
コントローラが配置されているブローカーID。現在のクラスタ内のコントローラがどのブローカーであるかが格納されます。
クラスタ内のすべてのブローカーに関する情報：各ブローカーのID、ラック情報、設定された複数の接続情報など
クラスター内のすべてのノードに関する情報: 厳密に言えば、前の項目と多少重複しますが、この項目はブローカー ID とリスナータイプ別にグループ化されています。非常に大規模なクラスターの場合、このキャッシュを使用すると、前の項目のコンテンツをトラバースせずに、指定されたノード情報をすばやく見つけることができるため、最適化されます。
クラスター内のすべてのパーティションに関する情報: いわゆるパーティション情報とは、パーティションのリーダー、ISR、AR 情報、および現在オフラインになっているレプリカのセットを指します。この部分のデータはトピック-パーティション ID に従ってグループ化されているため、各パーティションの現在のステータスをすばやく見つけることができます。 (注: AR は割り当てられたレプリカの略で、トピックの作成時にパーティションに割り当てられたレプリカのセットです)

ブローカー負荷分散

パーティション量の負荷: 各ブローカーのパーティション数は均一である必要があります。

パーティションのレプリカ割り当てアルゴリズムは次のとおりです。

すべてのブローカー（合計でn人のブローカーがいると仮定）と割り当てるパーティションを並べ替えます。
i番目のパーティションを(i mod n)番目のブローカーに割り当てる
i番目のパーティションのj番目のレプリカを((i + j) mod n)番目のブローカーに割り当てる

容量負荷: 各ブローカーが占有するハードディスク容量は均一である必要があります。

Kafka 1.1 より前のバージョンでは、Kafka は各ブローカーのパーティション数が均一であることを保証できましたが、各パーティション内のメッセージ数が異なるため、ハードディスク間でメモリ使用量に大きな違いが生じる可能性がありました。 Kafka1.1では、レプリカのクロスパス移行機能kafka-reassign-partitions.shが追加されました。監視システムと組み合わせることで、自動負荷分散を実現できます。

Kafka の高可用性

Kafkaの高可用性を紹介する前に、いくつかの概念を紹介しましょう。

同期レプリケーション: メッセージがコミットされたと見なされる前に、すべての作業中のフォロワーがレプリケーションを完了する必要があります。このレプリケーション方法はスループットレートに大きく影響します。
非同期レプリケーション: フォロワーはリーダーから非同期的にデータをプルします。データがリーダーによってログに書き込まれている限り、コミットされたとみなされます。この場合、フォロワーがリーダーより大幅に遅れ、リーダーが突然クラッシュすると、データが失われます。

イスル

Kafka は、ISR (パーティションリーダーと同期されたレプリカのリスト) を使用して同期レプリケーションと非同期レプリケーションを組み合わせ、データが失われないようにすることとスループットのバランスをとります。プロデューサーはパーティションリーダーにメッセージを送信するだけで、リーダーはメッセージをローカルログに書き込みます。フォロワーはリーダーからデータを取得します。メッセージを受信した後、フォロワーはリーダーに ACK を送信します。リーダーが ISR 内のすべてのレプリカから ACK を受信すると、メッセージはコミットされたとみなされ、リーダーは HW を増やしてプロデューサーに ACK を送信します。この方法では、リーダーに障害が発生した場合でも、Isr 内にレプリカが存在する限り、データは失われません。

ISR 動的更新

リーダーは ISR を追跡します。 ISR 内のフォロワーがクラッシュしたり、大幅に遅れたりした場合、リーダーはそれを ISR から削除します。ここで説明する「大幅に遅れる」とは、フォロワーによってコピーされたメッセージの数がリーダーより遅れて、事前設定された値 (replica.lag.max.messages) を超えるか、フォロワーが一定期間 (replica.lag.time.max.ms) リーダーにフェッチ要求を送信できないことを意味します。

Zookeeper のブローカーノード

/brokers/topics/[topic]/partitions/[partition]/state には、トピックパーティションのリーダーやIsrなどの情報が格納されます。

コントローラはブローカーの障害検出とフェイルオーバー（障害/回復）を担当します。

コントローラーは Zookeeper にウォッチを登録します。ブローカーがダウンすると、Zookeeper 内の対応する znode が自動的に削除されます。 Zookeeper はコントローラーによって登録されたウォッチをトリガーし、コントローラーは最新のブローカー情報を読み取ります。
コントローラーは、ダウンしているすべてのブローカー上のすべてのパーティションを含む set_p を決定します。
set_p 内の各パーティションに対して、新しいリーダーと Isr を選出し、結果を更新します。

3.1 /brokers/topics/[topic]/partitions/[partition]/stateからパーティションの現在のISRを読み取る

3.2 パーティションの新しいリーダーと Isr を決定します。現在の ISR 内に少なくとも 1 つのレプリカがまだ生きている場合は、そのうちの 1 つが新しいリーダーとして選択され、新しい ISR には現在の ISR 内に生きているレプリカがすべて含まれます。それ以外の場合は、パーティション内の生き残ったレプリカが新しいリーダーおよび ISR として選択されます (このシナリオでは、データが失われる可能性があります)

3.3 Leader、ISR、leader_epoch、controller_epochを更新: /brokers/topics/[topic]/partitions/[partition]/stateに書き込みます。

RPC を介して、set_p に関連するブローカーに LeaderAndISRRequest コマンドを直接送信します。コントローラーは、効率を向上させるために、1 つの RPC 操作で複数のコマンドを送信できます。

コントローラーがクラッシュする

各ブローカーは、Zookeeper の一時ノード「/controller」にウォッチャーを登録します。コントローラーがダウンすると、「/controller」が消え、ブローカーの監視がトリガーされます。各ブローカーは新しいコントローラーパスの作成を試み、そのうち 1 つだけが選出に成功し、コントローラーとして選出されます。

Kafka 使用時に冪等性を確保する方法

メッセージを失わないでください

まず、Kafka は少なくともコミットされたメッセージの配信を保証します。
送信者には再試行メカニズムがある
プロデューサービジネスパーティは、プロデューサーを使用してメッセージを送信するときに、コールバック関数を登録します。 onErrorメソッドでメッセージを再送信する
コンシューマーはメッセージをプルした後、それを処理し、コミットして、コミットされたメッセージが処理されたことを確認します。

繰り返しなし

コンシューマーはまずメッセージをプルして保存し、コミットが成功した後にキャッシュされたデータを削除します。

Kafka の高パフォーマンス