あなたは本当に Kafka のアーキテクチャ原理を理解していますか?

Apache Kafka は、LinkedIn によって最初に開発された分散メッセージング システムです。現在は Apache のサブプロジェクトであり、オープンソース分野で最も広く使用されているメッセージング システムの 1 つとなっています。

[[240583]]

Kafka コミュニティは非常に活発です。バージョン 0.9 以降、Kafka のスローガンは「高スループットの分散メッセージング システム」から「分散ストリーミング プラットフォーム」に変更されました。

Kafka は、以下の点で従来のメッセージング システムとは異なります。

• Kafka はスケールアウトが容易な分散システムです。
• パブリッシュとサブスクライブの両方で高いスループットを提供します。
• 複数のサブスクライバーをサポートし、障害が発生した場合にコンシューマーのバランスを自動的に調整します。
• メッセージの永続性。

Kafka と他のメッセージ キューの比較:

入門例

プロデューサー

コードは次のとおりです。

 java.util.Properties をインポートします。
 org.apache.kafka.clients.producer.KafkaProducer をインポートします。
 org.apache.kafka.clients.producer.ProducerRecord をインポートします。
パブリッククラス UserKafkaProducer は Thread を拡張します
{
    プライベート最終 KafkaProducer< Integer 、 String> プロデューサー;
    プライベート最終文字列トピック;
    プライベート最終プロパティ props = new Properties();
パブリックUserKafkaProducer(文字列トピック)
    {
        props.put( "metadata.broker.list" , "localhost:9092" );
        props.put( "bootstrap.servers" , "master2:6667" );
        props.put( "再試行" , 0);
        props.put( "バッチサイズ" , 16384);
        props.put( "linger.ms" , 1);
        props.put( "バッファメモリ" , 33554432);
        props.put( "key.serializer" 、 "org.apache.kafka.common.serialization.StringSerializer" );
        props.put( "value.serializer" 、 "org.apache.kafka.common.serialization.StringSerializer" );
        プロデューサー = 新しい KafkaProducer< Integer 、 String>(props);
        this.topic = トピック;
    }
 @オーバーライド
パブリックボイド実行（）{
 intメッセージ番号 = 1;
        （真）の間
        {
            文字列 messageStr = 新しい文字列 ( "Message_" + messageNo);
            システム。 out .println( "送信:" + messageStr);
            //非同期で送信されるFuture<RecordMetadata>を返します
            プロデューサー.send(新しいプロデューサーレコード< Integer 、String>(トピック、messageStr));
            メッセージ番号++;
            試す {
                スリープ(3000);
            } キャッチ (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
 }

消費者

コードは次のとおりです。

プロパティ props = new Properties();
 /* kakfa サービスのアドレスを定義します。すべてのブローカーを指定する必要はありません */
 props.put( "bootstrap.servers" , "localhost:9092" );
 /* コンシューマーグループを作成する*/
 props.put( "group.id" , "test" );
 /* オフセットを自動的に確認するかどうか */
 props.put( "enable.auto.commit" , "true" );
 /* オフセットの自動確認の時間間隔 */
 props.put( "auto.commit.interval.ms" , "1000" );
 props.put( "session.timeout.ms" , "30000" );
 /*キーシリアル化クラス */
 props.put( "key.deserializer" 、 "org.apache.kafka.common.serialization.StringDeserializer" );
 /* 値のシリアル化クラス */
 props.put( "value.deserializer" 、 "org.apache.kafka.common.serialization.StringDeserializer" );
 /* コンシューマーを定義する */
 KafkaConsumer<String, String> コンシューマー = 新しい KafkaConsumer<>(props);
 /* コンシューマーがサブスクライブするトピック。複数のトピックを同時にサブスクライブできます */
 consumer.subscribe(Arrays.asList( "foo" , "bar" ));
 /* データを読み取ります。読み取りタイムアウトは 100 ミリ秒です */
 （真）の間{
    ConsumerRecords<String, String> レコード = consumer.poll(100);
 ( ConsumerRecord<String, String> レコード: レコード)
        システム。 out .printf( "オフセット = %d、キー = %s、値 = %s" 、record.offset()、record.key ( )、record.value());
 }

Kafka アーキテクチャの原則

Kafka のアーキテクチャ原則に関して、まず次の質問をします。

• Kafka のトピックとパーティションは内部的にどのように保存されますか?それぞれの特徴は何ですか？
• 従来のメッセージング システムと比較した Kafka の消費モデルの利点は何ですか?
• Kafka はどのようにして分散データストレージとデータ読み取りを実現するのでしょうか?

Kafka アーキテクチャ図

カフカ用語集

Kafka アーキテクチャには、複数のプロデューサー、複数のブローカー、および複数のコンシューマーが存在します。各プロデューサーは複数のトピックに対応できますが、各コンシューマーは 1 つのコンシューマー グループにのみ対応できます。

Kafka アーキテクチャ全体は ZK クラスターに対応しており、クラスター構成を管理し、リーダーを選出し、コンシューマー グループが変更されたときに再バランスを実行します。

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

Kafka のすべてのメッセージにはトピックがあります。一般的に、アプリケーションで生成されるデータの種類ごとに異なるテーマを設定できます。

通常、トピックには複数のメッセージ サブスクライバーが存在します。プロデューサーがトピックにメッセージを公開すると、トピックをサブスクライブしているコンシューマーはプロデューサーによって書き込まれた新しいメッセージを受信できます。

Kafka はトピックごとに分散パーティション ログ ファイルを保持し、各パーティションは Kafka ストレージ レベルでの追加ログです。

このパーティションに公開されたメッセージはすべて、ログ ファイルの末尾に追加されます。パーティション内の各メッセージには、時系列順に単調に増加するシーケンス番号が割り当てられます。これがオフセットです。オフセットは Long 数値です。

このオフセットを使用して、このパーティション内の一意のメッセージを識別できます。パーティションでは順序が保証されますが、トピックでは保証されません。

上の図では、プロデューサーがどのパーティションに送信するかを決定します。

キー値がない場合、ポーリングが実行されます。

キー値がある場合は、キー値をハッシュし、パーティション数の係数を取得して、同じキー値を持つメッセージが同じパーティションにルーティングされるようにします。キューの強力な順次一貫性が必要な場合は、すべてのメッセージを同じキーに設定できます。

消費モデル

メッセージはプロデューサーから Kafka クラスターに送信された後、コンシューマーによって消費されます。一般的に言えば、消費モデルには 2 つあります。

• プッシュモデル
• プルモデル

メッセージ ブローカーが消費ステータスを記録する、プッシュ モデルに基づくメッセージ システム。メッセージ ブローカーはメッセージをコンシューマーにプッシュした後、メッセージを消費済みとしてマークしますが、この方法では消費の処理セマンティクスを適切に保証できません。

たとえば、コンシューマーにメッセージを送信した後、コンシューマー プロセスがクラッシュしたり、ネットワーク上の理由によりメッセージが受信されなかったりします。コンシューマー プロキシで消費済みとしてマークすると、メッセージは失われます。

プロデューサーがメッセージを受信した後に応答する方法を使用する場合、メッセージブローカーは消費ステータスを記録する必要があり、これは望ましくありません。

Push が使用される場合、メッセージの消費速度はコンシューマー エージェントによって完全に制御されます。消費者がブロックされると、問題が発生します。

Kafka は、消費速度と進行状況を自ら制御するプル モデル (Poll) を採用しています。消費者は任意のオフセットに応じて消費することができます。

たとえば、コンシューマーは、再処理のためにすでに消費されたメッセージを消費したり、最新のメッセージを消費したりすることができます。

ネットワークモデル

Kafka クライアント: シングルスレッドセレクター

シングルスレッド モードは、同時接続数が少なく、ロジックが単純で、データ量が少ない状況に適しています。 Kafka では、Consumer と Producer の両方が上記のシングルスレッド モードを使用します。

このモードは Kafka サーバーには適していません。サーバー上のリクエスト処理プロセスは比較的複雑であり、スレッドのブロックが発生します。後続のリクエストが発生すると、それらを処理することができなくなり、大量のリクエストがタイムアウトして雪崩が発生します。サーバーでは、実行ロジックを処理するためにマルチスレッドを最大限に活用する必要があります。

Kafka サーバー: マルチスレッド セレクター

Kafka サーバーはマルチスレッド セレクター モデルを使用します。アクセプターは別のスレッドで実行されます。読み取り操作用のスレッド プール内のすべてのスレッドは、セレクターに読み取りイベントを登録し、サーバーの読み取り要求のロジックを担当します。

読み取りが成功すると、要求はメッセージ キューの共有キューに配置されます。次に、書き込みスレッド プールで、要求を取り出して論理処理を実行します。

このように、リクエスト スレッドがブロックされた場合でも、メッセージ キューからリクエストを取得して処理する後続のスレッドが存在します。書き込みスレッドで論理処理が完了したら、OP_WIRTE イベントが登録されるので、それに応答を送信する必要があります。

信頼性の高い分散ストレージモデル

Kafka では、高信頼性モデルはレプリケーション メカニズムに依存しています。レプリケーションメカニズムにより、マシンがクラッシュしてもデータが失われることはありません。

高性能ログストレージ

Kafka トピックの下にあるすべてのメッセージは、パーティションの形式で複数のノードに分散して保存されます。

同時に、Kafka マシンでは、各パーティションは実際にはログ ディレクトリに対応しており、ディレクトリの下には複数のログ セグメント (LogSegment) が存在します。

LogSegment ファイルは、「.index」ファイルと「.log」ファイルの 2 つの部分で構成され、それぞれセグメント インデックス ファイルとデータ ファイルを表します。

これら 2 つのファイルのコマンド ルールは次のとおりです。グローバル パーティションの最初のセグメントは 0 から始まり、後続の各セグメント ファイル名は、前のセグメント ファイルの最後のメッセージのオフセット値になります。値は 64 ビット、20 桁の長さです。数字がない場合は0で埋められます。

以下のように、メッセージが 1000 個あり、各 LogSegment サイズが 100 であると仮定すると、900 ～ 1000 のインデックスとログは次のようになります。

Kafka メッセージ データが大きすぎるため、すべてのインデックスを作成すると、スペースが占有され、時間の消費が増加します。そのため、Kafka はスパース インデックス方式を選択し、インデックスをメモリに直接入力できるようにして、部分クエリの速度を向上させます。

データの読み取り方法について簡単に紹介します。 911 番目のデータを読み取る場合、最初のステップは、それがどのセクションに属しているかを見つけることです。

バイナリ検索を使用して、それが属するファイルを検索します。 0000900.index と 00000900.log を見つけたら、インデックスで (911-900) = 11 または 11 未満の最も近いインデックスを検索します。

ここでは、バイナリ検索を使用してインデックス[10,1367]を見つけます。次に、このインデックスの物理的な位置 1367 から遡って、911 個のデータが見つかるまで検索を続けます。

上記は特定のオフセットを見つけるプロセスについて説明していますが、ほとんどの場合、特定のオフセットを見つける必要はなく、順番に読み取るだけで済みます。

シーケンシャル読み取りでは、オペレーティング システムはメモリとディスクの間にページ キャッシュを追加します。これは通常行われる事前読み取り操作であるため、シーケンシャル読み取り操作は非常に高速になります。

しかし、Kafka には問題があります。パーティションが多すぎると、ログ セグメントも多数存在します。書き込む際は、一括して書き込むため、実際にはランダムな書き込みになります。この時点でランダム I/O はパフォーマンスに大きな影響を与えます。したがって、一般的に言えば、Kafka にはパーティションが多すぎることはできません。

この問題に対処するために、RocketMQ はすべてのログを 1 つのファイルに書き込み、順次書き込みに変換できます。特定の最適化により、読み取りを順次読み取りに近づけることもできます。

次のことを考えてみてください:

• なぜパーティションが必要なのでしょうか?つまり、トピックにパーティションが 1 つだけ存在することはできないのでしょうか?
• ログをセグメント化する必要があるのはなぜですか?

コピーメカニズム

Kafka のレプリケーション メカニズムでは、複数のサーバー ノードが他のノードのトピック パーティションのログを複製します。

クラスター内のノードに障害が発生すると、障害が発生したノードへのアクセス要求は他の正常なノードに転送されます (このプロセスは通常、Reblance と呼ばれます)。

各 Kafka トピックの各パーティションには、プライマリ コピーと 0 個以上のレプリカがあります。レプリカはプライマリ コピーとデータを同期させ、プライマリ コピーに障害が発生すると置き換えられます。

Kafka では、すべてのレプリカを使用してプライマリ レプリカを置き換えることはできないため、ISR (In Sync Replicas) セットが Kafka Leader ノードで維持されます。

同期セットとも呼ばれます。このセットでは、次の 2 つの条件を満たす必要があります。

• ノードは ZK との接続を維持する必要があります。
• 同期プロセス中、このレプリカはプライマリ レプリカから大幅に遅れることはできません。

レプリカの完全なセットを識別するための AR (割り当てられたレプリカ) もあり、遅れのために削除されたレプリカのセットを示すために OSR が使用されます。

したがって、式は次のようになります: ISR = リーダー + それほど遅れていないレプリカ。 AR = OSR + ISR です。

ここでは、2 つの用語について説明する必要があります。HW (ハイ ウォーター マーク) は、コンシューマーが確認できるこのパーティションの位置であり、LEO は各パーティションのログ内の最後のメッセージの位置です。

HW は、リーダーが配置されているブローカーに障害が発生した場合でも、メッセージが失われることなく、新しく選出されたリーダーからメッセージを取得できることを保証できます。

プロデューサーがリーダーにデータを送信するとき、データの信頼性のレベルは request.required.acks パラメータを通じて設定できます。

• 1 (デフォルト): これは、ISR のリーダーがデータを正常に受信し、確認を受け取った後にプロデューサーが次のメッセージを送信することを意味します。リーダーに障害が発生すると、データが失われます。
• 0: これは、プロデューサーがブローカーからの確認を待つ必要がなく、次のメッセージのバッチを送信し続けることを意味します。この場合、データ伝送効率は最大になりますが、データの信頼性も最大になります。
• -1: プロデューサーは、送信が完了したとみなされる前に、ISR 内のすべてのフォロワーがデータを受信したことを確認するまで待機する必要があります。信頼性は***です。

ただし、データが失われないことが保証されるわけではありません。たとえば、ISR にリーダーのみが存在する場合 (他のノードが ZK から切断されているか、追いついていない)、状況は acks = 1 になります。

高可用性モデルとべき等性

分散システムには、一般に 3 種類の処理セマンティクスがあります。

少なくとも一度

少なくとも 1 回、場合によっては複数回。プロデューサーが Ack から確認を受け取った場合、それはメッセージが Kafka に書き込まれたことを意味します。これは正確に 1 回であり、これを Exactly-once と呼びます。

ただし、プロデューサーがタイムアウトするかエラーを受信し、request.required.acks が -1 以外に設定されている場合は、メッセージの送信を再試行し、クライアントはメッセージが Kafka に書き込まれなかったと認識します。

ブローカーが Ack を送信する前に失敗したが、メッセージが Kafka に正常に書き込まれた後の場合、この再試行によりメッセージが 2 回書き込まれることになります。

したがって、メッセージは最終コンシューマーに複数回配信され、コンシューマー処理ロジックが冪等性を保証しない場合は誤った結果が得られます。

このセマンティクスでは、混乱が発生する可能性があります。つまり、最初の Ack が失敗し、再試行の準備ができているが、2 番目のメッセージがすでに送信されている場合、単一のパーティションで混乱が発生する可能性があります。

Producer パラメータ max.in.flight.requests.per.connection を設定する必要があります。 Flight.requests は、プロデューサーが応答のない送信されたリクエストを保存するために使用するキューであり、プロデューサー側で応答のないリクエストの数が 1 であることを保証します。

最大1回

Ack がタイムアウトになったりエラーが返されたりしたときに Producer が再試行しない場合、つまり request.required.acks = -1 とすると、最終的にメッセージが Kafka に書き込まれず、Consumer はメッセージを受信しません。

正確に1回

プロデューサーがメッセージの送信を再試行した場合でも、メッセージは最大で 1 回コンシューマーに配信されることが保証されます。このセマンティクスは最も理想的ですが、実現するのが最も困難でもあります。

0.10 より前では、正確に 1 回は保証されず、コンシューマーによって提供されるべき等性の保証が必要でした。 0.11.0 では、トランザクションを使用してこれを保証します。

正確に1回だけを実装する方法

正確に 1 回だけ実行するために、Kafka 0.11.0 には 2 つの公式戦略があります。

単一プロデューサー 単一トピック

各プロデューサーは初期化時に一意の PID が割り当てられます。それぞれの一意の PID について、プロデューサーから指定されたトピック内の特定のパーティションに送信されるメッセージには、0 から単調に増加するシーケンス番号が付けられます。

また、ブローカー側でもディメンションを維持し、メッセージが送信されるたびにそれを調整して検証します。

• メッセージのシーケンス番号がブローカーが保持するシーケンス番号より 1 以上大きい場合は、途中に書き込まれていないデータがある、つまり順序​​が間違っていることを意味します。このとき、ブローカーはメッセージを拒否し、プロデューサーは InvalidSequenceNumber をスローします。
• メッセージのシーケンス番号がブローカーによって保持されるシーケンス番号以下の場合、メッセージが保存されていること、つまり重複メッセージであることを意味します。ブローカーはメッセージを直接破棄し、プロデューサーは DuplicateSequenceNumber をスローします。
• メッセージのシーケンス番号がちょうど 1 大きい場合、正当であることが証明されます。

上記により、次の 2 つの問題が解決されます。

• プロデューサーがメッセージを送信した後に失敗すると、ブローカーはそれを保存しませんが、2 番目のメッセージは正常に送信され、データの乱れが発生します。
• プロデューサーがメッセージを送信した後、ブローカーはそれを正常に保存しますが、Ack の返信に失敗するため、プロデューサーは重複したメッセージを再度配信します。

上記のすべては同じ PID の下にあるため、単一のプロデューサー内の同じセッション内にあることが保証される必要があります。プロデューサーが切断して新しい PID が割り当てられる場合、これは保証できないため、Kafka にはこれを保証するためのトランザクション メカニズムがあります。

取引

Kafka におけるトランザクションの役割は次のとおりです。

• 正確に 1 回のセマンティクスを実装します。
• 操作のアトミック性を確保し、すべてが成功するか、すべてが失敗するようにします。
• ステートフル操作の回復。

トランザクションでは、複数のキューにまたがる場合でも、このトランザクション内のコンシューマー キューに対する操作がアトミックであるとみなされ、すべてが成功するか、すべてが失敗するかのいずれかになります。

さらに、ステートフル アプリケーションでは、再起動後にブレークポイントから処理が継続されること、つまりトランザクションの回復も保証できます。

Kafka トランザクションでは、アプリケーションはクラッシュや再起動後も変更されない一意のトランザクション ID (トランザクション ID) を提供する必要があります。

トランザクション ID と PID は 1 対 1 で対応している場合があります。違いは、トランザクション ID はユーザーによって提供されるのに対し、PID はユーザーにとって透過的な内部実装である点です。

プロデューサーが再起動された後に同じトランザクション ID を持つ古いプロデューサーを無効にするために、プロデューサーがトランザクション ID を通じて PID を取得するたびに、単調に増加するエポックも取得します。

古いプロデューサーのエポックは新しいプロデューサーのエポックよりも小さいため、Kafka はプロデューサーが古いことを簡単に識別し、その要求を拒否できます。

これを実現するために、Kafka 0.11.0.0 では、プロデューサーによって送信されたメッセージのトランザクション性を管理する Transaction Coordinator と呼ばれるサーバー側モジュールが導入されました。

トランザクション コーディネーターは、内部トピックに保存されているトランザクション ログを管理します。

トピック データは永続的であるため、トランザクションの状態も永続的です。プロデューサーはトランザクション ログを直接読み書きしません。代わりに、トランザクション コーディネータと通信し、トランザクション コーディネータがトランザクション ステータスを対応するトランザクション ログに挿入します。

トランザクション ログの設計は、コンシューマーのオフセットを格納するために使用されるオフセット ログの設計と似ています。

***

メッセージ キューや Kafka に関する一般的な面接の質問に関して、上記の記事では次のような典型的な質問を抽出できます。上記の要約を読めば、ほとんどの質問に答えることができます。

• メッセージ キューを使用する理由は何ですか?メッセージキューの役割は何ですか?
• Kafka のトピックとパーティションは内部的にどのように保存されますか?それぞれの特徴は何ですか？
• 従来のメッセージング システムと比較した Kafka の消費モデルの利点は何ですか?
• Kafka はどのようにして分散データストレージとデータ読み取りを実現するのでしょうか?
• Kafka が RocketMQ よりも少ないスタンドアロン パーティションをサポートするのはなぜですか?
• なぜパーティションが必要なのでしょうか?つまり、トピックにパーティションが 1 つだけ存在することはできないのでしょうか?
• ログをセグメント化する必要があるのはなぜですか?
• Kafka は高い信頼性と高可用性を維持するためにどのようなメカニズムに依存していますか?
• メッセージ キューはどのようにしてメッセージの冪等性を保証するのでしょうか?
• 自分でメッセージ キューを設計するように求められた場合、どのように設計し、どのような点を考慮しますか?

著者: 李趙

自己紹介: 現在、Meituan Dianping ケータリング エコシステム技術部門で働いており、オープン ソース コードを勉強したり読んだりするのが好きです。 3 年以上の経験を持つ Java 開発者を募集しています。履歴書を [email protected] までお送りください。