さようなら、カフカ！さようなら、RocketMQ!

Pulsar は、2016 年に Yahoo によってオープンソース化され、2018 年にトップレベルの Apache プロジェクトとなったメッセージ ミドルウェアです。

[[377358]]

画像はPexelsより

オープンソース業界にはすでに多くのメッセージ キュー ミドルウェアが存在します。新しい力としてのパルサーの利点は何ですか?

Pulsar は誕生以来、他のメッセージ キュー (Kafka、RocketMQ など) と常に比較されてきました。

ただし、Pulsar の設計コンセプトは、ほとんどのメッセージ キュー ミドルウェアとは異なります。高スループット、低レイテンシ、コンピューティングとストレージの分離、マルチテナント、リモートレプリケーションなどの機能を備えています。

そのため、Pulsar は次世代のメッセージ キュー ミドルウェアとしても知られています。次に、一つずつ詳しく分析していきます。

パルサーアーキテクチャの原則

Pulsar アーキテクチャの原理は次のとおりです。

全体的なアーキテクチャは、他のメッセージ キュー ミドルウェアとそれほど変わりません。皆さんもよくご存知の用語を数多く目にしたことがあると思います。次に、これらの用語の意味を一つずつ説明していきます。

用語集:

• プロデューサー: メッセージ プロデューサー。ブローカーにメッセージを送信します。
• コンシューマー: メッセージ コンシューマー。消費処理のためにブローカーからクライアントへのメッセージを読み取ります。
• Broker: Pulsar のサーバーとみなすことができます。プロデューサーとコンシューマーは、クライアント メッセージ処理のノードと見なされます。

Pulsar のブローカーは他のメッセージ ミドルウェアとは異なります。ステートレスでストレージもないので、無制限に拡張できます。これについては後で詳しく説明します。

• Bookie: Apache Bookeeper を使用して、すべてのメッセージの永続性を維持します。
• ZK: Kafka と同様に、Pulsar も構成管理、トピックの割り当て、テナントなどのメタデータを保存するために ZK を使用します。
• サービスディスカバリ: Pulsar の Nginx として理解できます。たった 1 つの URL でブローカー全体と対話できます。もちろん、独自のサービス検出を使用することもできます。

トピックの読み取り、更新、または削除のためにクライアントによって行われた最初の要求は、そのトピックを処理しているブローカーではないブローカーに送信されることがあります。

このブローカーがトピックの要求を処理できない場合、ブローカーはトピック要求を処理できるブローカーに要求をリダイレクトします。

Kafka、RocketMQ、または Pulsar のいずれであっても、メッセージ キュー ミドルウェアの最も重要な部分は、おそらく次の 3 つの部分に分かれます。

• プロデューサーがメッセージを生成し、対応するブローカーに送信する方法。
• ブローカーがメッセージを処理し、効率的に保存し、クエリを実行する方法。
• コンシューマーはメッセージを消費する方法です。

これら3つの部分についても後ほど説明します。

プロデューサーはメッセージを生成する

コードを使用してメッセージを送信する方法を簡単に見てみましょう。

 PulsarClient クライアント = PulsarClient。 ( "pulsar://pulsar.us-west.example.com:6650" )を作成します。 
 
プロデューサー プロデューサー = client.createProducer(
 "persistent://sample/standalone/ns1/my-topic" ); 
 
 //トピックに10件のメッセージを公開する
( int i = 0; i < 10; i++) {
    プロデューサー.send( "私のメッセージ" .getBytes());
 }

ステップ 1: まず、URL を使用してクライアントを作成します。この URL は、サービス ディスカバリのアドレスです。スタンドアロンモードを使用すると、直接接続できます。

ステップ 2: URL に似たパラメータを渡しました。このパラメータを渡す必要があるのは、それを作成したトピックまたは名前空間を指定する場合だけです。 URL の形式は次のとおりです。

 {永続|非永続}://テナント/名前空間/トピック

ステップ 3: Send メソッドを呼び出してメッセージを送信します。非同期送信をサポートするために、sendAsync メソッドも提供されています。

上記の 3 つのステップのうち、ステップ 1 と 2 は準備段階に属し、クライアントとプロデューサーを構築するために使用されます。私たちのコアロジックは Send にあります。

ここでいくつか小さな疑問を提起したいと思います。まず、他のメッセージ キューでどのように行われるかを考えてから、それを Pulsar と比較することができます。

• Send を呼び出すと、すぐに送信されますか?
• パーティションが複数ある場合、どのブローカーに送信すればよいかをどのように見つければよいですか?

送信モード

上記で Send は Async モードと Sync モードに分かれていると説明しましたが、実際には Pulsar 内の Sync モードでも Async モードが採用されています。同期モードでは、同期効果を実現するためにコールバック ブロッキングがシミュレートされます。

このモードは Kafka でも採用されていますが、RocketMQ ではすべての送信が完全に同期されており、ブローカーに直接要求されます。

このモデルに基づいて、Pulsar と Kafka ではバッチ送信がサポートされ、RocketMQ では直接送信がサポートされます。一括送信のメリットは何ですか?

送信する TPS が特に高い場合、送信するたびにブローカーに直接接続すると、圧縮、認証、リンク作成などの多くの繰り返し作業が必要になる可能性があります。

たとえば、1,000 件のメッセージを送信する場合、この繰り返し作業を 1,000 回実行する必要がある可能性があります。バッチで送信すると、これらの 1,000 件のメッセージが 1 つのリクエストに結合され、比較的圧縮されるため、認証作業は 1 回だけ実行すれば済みます。

学生の中には、一括送信によって送信時間に一定の遅延が生じるのかと疑問に思う人もいるかもしれません。実際のところ、これについて心配する必要はありません。 Pulsar では、デフォルトのスケジュールでは 1 ミリ秒ごとにバッチが送信されます。または、バッチサイズがデフォルトで 1000 に達すると、バッチが送信されます。送信頻度は依然として非常に高速です。

送信負荷分散

メッセージ キューでは、トピックは通常、水平方向に展開されます。 Pulsar と Kafka では、これはパーティションと呼ばれ、RocketMQ ではキューと呼ばれます。本質的にはそれらはすべてパーティションです。異なるブローカーに異なるパーティションを配置することで、水平拡張の効果を実現できます。

送信するときに、パーティションを選択するための独自の戦略を作成することも、デフォルトのラウンドロビン パーティション戦略を使用することもできます。

パーティションを選択した後、どのパーティションがどのブローカーに対応するかをどのように判断するのでしょうか?

まず次の図を見てください。

ステップ 1: すべての情報パーティション マッピング情報は、ZK と Broker のキャッシュに保存されます。

ステップ 2: ブローカーにクエリを実行することで、パーティションとブローカーの関係を取得し、定期的に更新できます。

ステップ 3: Pulsar では、各パーティションは送信側で個別のプロデューサーに抽象化されます。これは、Kafka や RocketMQ とは異なります。

Kafka では、パーティションを選択した後、パーティションに対応するブローカー アドレスを見つけて送信します。

Pulsar は各パーティションをプロデューサーにカプセル化します。コード実装では、どのブローカーに対応するかを気にする必要はありません。すべてのロジックは Producer コード内にあり、全体的には比較的クリーンです。

圧縮されたメッセージ

メッセージ圧縮は、情報伝送を最適化する手段の 1 つです。通常、大きなファイルは圧縮パッケージの形式でダウンロード用に提供されます。

このアイデアはメッセージ キューでも使用できます。たとえば、1,000 メッセージのバッチの送信サイズは 1 MB ですが、圧縮後は数十 KB になる可能性があり、これによりブローカー間の送信効率は向上しますが、同時に CPU の損失も発生します。

Pulsar クライアントは、lz4、zlib、zstd、snappy などの複数の圧縮タイプをサポートしています。

クライアント.newProducer()
    .topic("テストトピック")
    .compressionType(圧縮タイプ.LZ4)
    。作成する（）;

ブローカ

次に、2 番目に重要な部分であるブローカーについて説明しましょう。 Broker の設計において、Pulsar は他のすべてのメッセージ キューとはまったく異なっており、まさにこの違いが Pulsar の特徴となっています。

コンピューティングとストレージの分離

まず、最大の特徴であるコンピューティングとストレージの分離についてお話しします。

冒頭で述べたように、Pulsar は次世代のメッセージ キューであり、そのアーキテクチャ設計から大きな恩恵を受けています。 Kafka でも RocketMQ でも、すべてのコンピューティングとストレージは同じマシンに配置されます。

このモデルにはいくつかの欠点があります。

• 拡張の難しさ: クラスターを拡張する必要がある場合、通常は CPU またはディスクの問題が原因です。しかし、CPU とディスクの構成が適切である可能性のあるマシンを適用する必要があり、結果としてリソースが無駄になります。さらに、Kafka の拡張にはデータ移行も必要となり、これは非常に複雑なプロセスです。
• 不均衡な負荷: 一部のパーティションに大量のデータがある場合、ブローカーに不均衡な負荷が発生します。下の図に示すように、パーティションに大量のデータがある場合、ブローカー (船) は大量のデータを運びますが、他のブローカーは比較的アイドル状態になる可能性があります。

Pulsar のコンピューティング分離アーキテクチャは、この問題を非常にうまく解決できます。

• コンピューティングの場合: メッセージ キューの読み取りと書き込みを提供するブローカーは、データを保存せず、ステートレスであるため、拡張に非常に適しています。十分な数のマシンがあれば、自由に使用できます。

ブローカーの容量をスケーリングすることは、コンシューマーのスループットを向上させるためによく使用されます。電子商取引のプロモーションなど、トラフィック量の多いビジネスやアクティビティがある場合は、事前にブローカーの容量を拡張できます。

• ストレージの場合: メッセージ キューのストレージのみを提供する Bookie を使用します。メッセージ量の要件がある場合は、Bookie を拡張することができ、データを移行する必要がないため、拡張は非常に便利です。

メッセージストレージ

名詞分析:

上の図は、Bookie の読み取り/書き込みアーキテクチャの図です。最初に紹介する必要がある用語がいくつかあります。

• エントリ: エントリ ID、レコード エンティティなどを含む、ブックキーパーに保存されるレコードです。
• 元帳: 元帳はエントリを保存するために使用するものと考えられており、複数のエントリ シーケンスが元帳を構成します。
• ジャーナル: 実際には、ブックキーパーのトランザクション ログを保存するために使用されるブックキーパーの WAL (先行書き込みログ) です。ジャーナル ファイルには最大サイズがあります。このサイズに達すると、新しいジャーナル ファイルが作成されます。
• エントリ ログ: エントリを保存するファイル。元帳は論理的な概念です。エントリは元帳ごとに集計され、エントリ ログ ファイルに書き込まれます。同様に、エントリ ログにも最大値があり、最大値に達すると新しいエントリ ログ ファイルが作成されます。
• インデックス ファイル: 元帳のインデックス ファイル。元帳のエントリはエントリ ログ ファイルに書き込まれます。インデックス ファイルは、エントリ ログ ファイル内の各元帳にインデックスを付けるために使用され、エントリ ログ内の各元帳の保存場所とエントリ ログ ファイル内のデータの長さを記録します。
• メタデータ ストレージ: メタデータ ストレージは、ブッキー上の元帳など、ブッキー関連のメタデータを保存するために使用されます。 Bookkeeper は現在 zk ストレージを使用しているため、bookkeeper をデプロイする前に、まず zk クラスターを用意する必要があります。

全体的なアーキテクチャの作成プロセスは次のとおりです。

• ステップ 1: ブローカーは書き込み要求を開始し、最初に WAL をジャーナル ディスクに書き込みます。 MySQL に詳しい友人は redolog を知っています。ジャーナルと redolog は、非永続的なデータを回復するという同じ機能を持ちます。
• ステップ 2: 次に、データをインデックスと元帳に書き込みます。パフォーマンスを維持するために、データはディスクに直接書き込まれるのではなく、ページ キャッシュに書き込まれ、その後非同期的にディスクにフラッシュされます。
• ステップ 3: 書き込みを確認します。

読み取りプロセスは次のとおりです。

• ステップ 1: 当然ながら、まずインデックスを読み取り、次にキャッシュを読み取り、次にディスクに移動します。
• ステップ 2: インデックスを取得したら、エントリ ロガーに移動して、インデックスに従って対応するデータを取得します。

効率的に読み書きするにはどうすればいいでしょうか? Kafka では、トピックが増えると、トピックごとに 1 つのファイルがあるため、ディスク IO が順次書き込みからランダム書き込みに変わります。

RocketMQ では、複数のトピックが 1 つの書き込みファイルに対応し、書き込みが順次行われるため、読み取りによってページキャッシュが簡単に上書きされ、更新され、IO に大きな影響を与える可能性があります。

そのため、Pulsar はこれらの問題に対処するために、読み取りと書き込みの両方で多くの最適化を行いました。

書き込みプロセス: 順次書き込み + ページキャッシュ。書き込みプロセスでは、すべてのファイルは独立したディスク上にあり、ジャーナルのみが同期的にフラッシュされます。

ジャーナルは journal-wal ファイルを順次書き込み、順次書き込みの効率は非常に高くなります。元帳とインデックスの両方に複数のファイルがありますが、Pagecache に書き込み、非同期でディスクにフラッシュするだけなので、ランダム書き込みはパフォーマンスに影響しません。

読み取りプロセス: ブローカー キャッシュ + ブッキー キャッシュ。 Pulsar はテーリング読み取りに非常に適しており、基本的に IO を使用しません。

一般的に言えば、コンシューマーはプロデューサーから送信されたメッセージをすぐに受け取るため、この部分は永続化後もブローカー内にキャッシュとして残ります。

もちろん、ブローカーにキャッシュがない場合でも (たとえば、ブローカーが新しく作成された場合)、Bookie は Memtable 内に独自のキャッシュを持つため、複数のキャッシュを介した読み取りプロセスの IO が削減されます。

最も理想的なケースでは、読み取りと書き込みの IO が完全に分離されているため、Pulsar で何百万ものトピックをサポートするのは簡単ですが、これは Kafka と RocketMQ では非常に困難です。

無制限のストリーミングストレージ

トピックは、実際には元帳ストリーム (セグメント) です。この設計により、Pulsar は単なる単純なメッセージ キュー システムではなくなります。ストリーミングシステムを置き換えることもできるため、Flink などのシステムを置き換えることができるストリームネイティブ プラットフォームとも呼ばれます。

イベント ストリーム (トピック/パーティション) は複数のセグメント ストレージで構成されており、各セグメントはエントリで構成されていることがわかります。これは、送信するメッセージの各バッチが通常エントリとして扱われるためと考えられます。

セグメントは、ファイルを書き込むための基本的な次元と見なすことができます。同じセグメントのデータは同じファイルに書き込まれ、異なるセグメントは異なるファイルに書き込まれ、セグメント間のデータはメタデータに保存されます。

階層型ストレージ

Kafka と RocketMQ では、ディスクの容量制限があるため、メッセージは一定期間保存されます。

この機能は Pulsar でも提供されていますが、メッセージを永続的に保存したい場合は階層型ストレージを使用できます。古いデータを定期的に S3 などの安価なストレージに更新することで、メッセージ キューを無期限に保存することができます。

データ複製

Pulsar のデータレプリケーションは、Kafka や RocketMQ とは大きく異なります。他のメッセージ キューでは、通常、他のレプリカが同期を主導するため、時間が予測不可能になります。

Pulsar は Qurom に似たプロトコルを使用します。これは、利用可能な Bookie プールのグループを提供し、そのうちのいくつかが正常に返される限り (通常は 1/2 以上)、それらのいくつかに同時に書き込みます。

• アンサンブル サイズ (E): 特定の元帳で使用できるブッキー プールのサイズを決定します。
• 書き込みクォーラム サイズ (Qw): Pulsar がエントリを書き込む Bookie の数を指定します。
• Ack クォーラム サイズ (Qa): Ack 書き込みが必要な Bookie の数を指定します。

この同時書き込み方式を使用すると、特にデータのコピーが多数ある場合に、データのレプリケーションがより効率的になります。

消費者

次に、Pulsar の最後の重要なコンポーネントである Consumer について説明します。

サブスクリプションモデル

サブスクリプション モードは、メッセージがさまざまなコンシューマーにどのように配信されるかを定義するために使用されます。さまざまなメッセージ キュー ミドルウェアには、独自のサブスクリプション モードがあります。

一般的に、当社の一般的なサブスクリプション モデルは次のとおりです。

• クラスター モード: メッセージはクラスター内のコンシューマーによってのみ消費されます。
• ブロードキャスト モード: メッセージはクラスター内のすべてのコンシューマーが使用できます。

Pulsar では 4 つのサブスクリプション モードが提供されています。

排他的: 名前が示すように、1 人の消費者のみが使用できます。 2 番目のコンシューマーが同じクラスターに登録されると、2 番目のコンシューマーは失敗します。これは、グローバルに順序付けられたメッセージに適用されます。

災害復旧：専用の拡張バージョン。排他的な接続が切断された場合、自動的に別の適切なコンシューマーに切り替わりますが、排他的な接続は 1 つしか存在できません。

共有モード: このモードはクラスター モードに少し似ています。メッセージは 1 つのクラスター内のコンシューマーによってのみ消費されます。ただし、RocketMQ とは異なり、RocketMQ はパーティション ディメンションに基づいています。同じパーティション内のデータは 1 台のマシンに送信されます。

Pulsar では、消費はパーティション ディメンションに基づいていませんが、すべてのコンシューマーがポーリングされてメッセージが送信されます。これの利点は何ですか?

マシンが 100 台あってもパーティションが 10 個しかない場合、実際に実行できるコンシューマーは 10 個だけですが、Pulsar では 100 台すべてのマシンがコンシューム処理を実行できます。

キー共有: 上記のパーティション ディメンションと同様に、RocketMQ では、同じキーを持つ連続メッセージが 1 つのパーティションに送信されます。

ただし、ここにはパーティション ディメンションはありません。代わりに、データはキーのハッシュに基づいて固定のコンシューマーに割り当てられ、コンシューマーの容量がパーティションの数に制限されるという問題も解決されます。

メッセージ取得モード

Kafka でも RocketMQ でも、クライアントは定期的にブローカーをポーリングしてメッセージを取得します。このモードはロングポーリングモードと呼ばれます。

このモードには、ネットワークのオーバーヘッドが比較的大きいという欠点があります。 Consumer が消費されるまでの遅延を計算してみましょう。ブローカーとコンシューマー間のネットワーク遅延は R であると仮定します。

合計時間は次のようになります。

• メッセージ A がブローカーに到着すると、ロングポーリングによってデータがパッケージ化されて返されます。ブローカーが消費者に戻るまでにかかる時間は R です。
• コンシューマーは再度リクエストを送信します。今回は R です。
• メッセージ A を Consumer (再び R) に返します。

ネットワーク遅延だけを考慮すると、メッセージの消費遅延は約 3R であることがわかります。そのため、これを最適化するための方法を考える必要があります。

メッセージが届いたら、それを直接消費者にプッシュするのが正しいのではないだろうかとすぐに考える学生もいるかもしれません。この方法では、遅延は R に 1 回だけになります。これが一般的なプッシュ モードです。

ただし、単純なプッシュ モードには問題があります。生産速度が消費速度よりもはるかに速い場合、プッシュされたメッセージによってメモリが確実に爆発的に増加します。これはバックプレッシャーです。

では、バックプレッシャーをどうやって解決するのでしょうか?プッシュ メソッドを最適化し、動的プッシュに変換できます。ロングポーリングを組み合わせ、ロングポーリング要求が行われたときに残りのバッファスペースをブローカーに通知します。ブローカーはデータをプッシュする責任があります。

この時点で、ブローカーはプッシュできるデータの最大数を把握しているため、コンシューマーに負担がかからないようにプッシュ動作を制御できます。

たとえば、コンシューマーがリクエストを開始すると、バッファーの残り容量は 100 になり、ブローカーは毎回最大 32 個のメッセージを返します。

次に、コンシューマーのこのロングポーリング要求に対して、ブローカーは 3 つのプッシュ (合計 96 メッセージ) を実行し、コンシューマーに応答を返します (応答には 4 つのメッセージが含まれます)。

ロングポーリング モデルが使用される場合、ブローカーはコンシューマーがリクエストを送信するたびに応答を実行します。

この例では、4 つのロングポーリング インタラクション (4 つのリクエストと 4 つのレスポンス、合計 8 つのネットワーク操作) が必要です。動的プッシュ/プルでは、​​1 つのリクエスト、3 つのプッシュ、1 つのレスポンス、合計 5 つのネットワーク操作が必要です。

したがって、Pulsar はこのメッセージ取得モードを採用して、コンシューマー層からのメッセージ到着時間をさらに最適化します。

この設計は非常に巧妙であり、多くのミドルウェアのロングポーリングモードはこのアイデアに基づいて改善できると思います。

要約する

Apache Pulsar の設計思想の多くは他のミドルウェアとは異なりますが、間違いなく未来に近いものです。

将来的には、他のメッセージ ミドルウェアの開発もこれに近づくだろうという大胆な予測です。現在、中国ではPulsarユーザーがますます増えています。 Tencent Cloud は Pulsar のクラウド バージョンである TDMQ を提供します。

もちろん、Huawei、Zhihu、Huyaなど、他の有名企業も徐々にこれを試みています。 Pulsar は本当にトレンドだと思います。

最後に、最近のビッグ・リバーの最終回のセリフを思い出しました。

あらゆる変化には苦痛と回り道が伴う可能性があり、開かれた道は広く平坦なものではないでしょう。しかし、大河が勢いよく流れていく流れは、危険な浅瀬や岩礁によってさえぎられることはない。道があるところなら、たとえ何万人もの反対者がいても、私は行きます。

著者: コーヒーラテ

編集者：タオ・ジアロン

出典：公式アカウント「Coffee Latte」（ID: close_3092860495）より転載