RocketMQ プロデューサーにこれほど多くの用途があることをなぜ知らなかったのでしょうか?

[[383770]]

この記事はWeChatの公開アカウント「大宇賢人」から転載したもので、著者は大宇です。この記事を転載する場合は、大悠仙人の公開アカウントにご連絡ください。

序文

メッセージ キュー RocketMQ は、Apache RocketMQ をベースに Alibaba Cloud が構築した、低レイテンシ、高同時実行性、高可用性、高信頼性を備えた分散メッセージ ミドルウェアです。

私の以前の記事を読んだことがある人は、メッセージ キューの概要を理解しているはずです。それで、このメッセージはどこから来たのでしょうか?

黄河の水は天から降ってくると言われています。自然界のあらゆるものは、理由もなく存在するわけではありません。それはどこから来たのですか?それは母親によって生産されます。シャネルはどこから来たのですか？機械と原材料によって生産されます。私たちが食べるお米にも、その由来があります。私たちはどこから来たのでしょうか？もちろん、私たちは偉大な母によって生み出されたのです。

ところで、あなたの偉大なお母さんに感謝し、日曜日に電話するのを忘れないでください

本題に入りましょう。これが境界線です。

RocketMQ バージョンのメッセージ キューは、分散アプリケーション システムに非同期分離とピークシフト機能を提供できるだけでなく、大量のメッセージの蓄積、高いスループット、信頼性の高い再試行など、インターネット アプリケーションに必要な機能も備えています。以下に機能の一部を挙げます

• メッセージ クエリ: RocketMQ には、メッセージ ID、メッセージ キー、トピックによるクエリの 3 つのメッセージ クエリ メソッドが用意されています。
• クエリ メッセージ トレース: メッセージ トレースを使用すると、メッセージ キュー RocketMQ サーバーを介してメッセージ プロデューサーからメッセージ コンシューマーまでの完全なリンクを明確に特定できるため、問題の特定とトラブルシューティングに便利です。
• クラスター消費とブロードキャスト消費: クラスター消費モードを使用する場合、RocketMQ バージョンのメッセージ キューは、すべてのメッセージはコンシューマー クラスター内の任意のコンシューマーによってのみ処理される必要があると認識します。ブロードキャスト消費モードを使用する場合、RocketMQ バージョンのメッセージ キューは、各メッセージをコンシューマー クラスター内のすべての登録済みコンシューマーにプッシュし、各マシンでメッセージが少なくとも 1 回消費されるようにします。
• 消費位置をリセット: 時間や位置に応じて消費の進行状況をリセットし、ユーザーがメッセージをさかのぼったり、蓄積されたメッセージを破棄したりできるようにします。
• デッドレターキュー: 正常に処理できないメッセージを、後続の処理のために特別なデッドレターキューに保存します。
• グローバル情報ルーティング: 世界中のさまざまな地域間でのメッセージ同期に使用され、地域間のデータの一貫性を確保します。

クライアントは実は非常に理解しやすいです。 RocketMQ はメッセージ サービスとして考えることができます。これはサービスなので、このサービスを呼び出す必要があります。では、このサービスを呼び出すと、メッセージはどこから来るのでしょうか?これはビジネスシナリオによって異なります。したがって、メッセージ プロデューサーはクライアントに属します。メッセージが生成されたら、それを永久にそこに残しておくことはできません。誰かがこれらのメッセージを処理する必要があります。これもビジネスによって決定されるため、メッセージの消費者もクライアントに属します。

次に、Dayu がこのクライアントの有用性について説明します。

プロデューサー

プロデューサーは、その名前が示すように、メッセージを生成する責任を負います。この時点で、プロデューサーはどこにメッセージを送信するのか、プロセスは何か、どのような種類のメッセージが送信されるのかなど、多くの疑問が頭に浮かぶはずです。これらの疑問を理解できれば、プロデューサー クライアントは基本的に完成です。

ゆうゆがみんなにちょっとしたコツを教えます。何かを学ぶには、まず全体的なプロセスを理解し、次にそれを分解して詳細に取り組み、最後に全体を理解します。これは非常に良い効果をもたらすでしょう。これは独占秘密レシピです。

次に、メッセージの送信方法 (負荷分散、フォールト トレランス メカニズム)、メッセージの送信先と保存場所、メッセージの種類という 3 つの側面から Producer を紹介します。

1. メッセージはどのように送信されますか?

まず第一に、メッセージが生成されてもどこにも送信されないということはあり得ません。そうすると、このメッセージを生成する意味がなくなるからです。したがって、このメッセージは常にどこかに送信され、中継される必要があります。次の図を参照してください。

プロデューサーはまず、指定されたトピックをローカル キャッシュから取得します。見つかった場合は、このトピックに基づいて生成されたメッセージを直接送信します。ご存知のとおり、キャッシュは速度を最適化し、ネットワーク転送を削減するためのものです。

そうでない場合は、ネーム サーバーにアクセスして最新のトピック リスト (ブローカーの起動時にネーム サーバーに登録されます) を取得し、特定の戦略を通じて MessageQueue キューを選択し、この MQ が配置されているブローカー アドレスを取得し、最初にローカル キャッシュからも取得する必要があります。取得できない場合は、ネームサーバに取得を依頼し（ブローカーアドレスとトピックのマッピング関係もネームサーバに登録されている）、メッセージを送信する。

送信に失敗した場合は再試行メカニズムがあり、デフォルトでは3回再試行されます。

実際、これほどの節約は、NameServer と NameServer 間のネットワーク転送を削減できるだけでなく、NameServer への負荷も軽減します。 NameServer 自体は軽量設計であるため、NameServer への負荷を軽減するのにも役立ちます。 NameServer については別途記事で紹介する予定です。

負荷分散

メッセージを送信するときには、まず対応するトピックが選択されることがわかります。各トピックは複数の MessageQueue に対応します。これは問題を引き起こします。メッセージを均等に送信できない場合、一部のキューには多くのメッセージが含まれ、一部のキューには少ないメッセージが含まれる可能性があり、リソースの無駄が発生します。

RocketMQ は、ポーリングという単純な方法を使用します。高級食材は、たいてい最もシンプルな調理法で十分です～

プロデューサーは、トピックの下にあるすべての MessageQueues をポーリングすることにより、送信側で負荷分散を実現します。簡単に言えば、次の図に示すように、誰もがシェアを持っています。

このようにして、トピックのメッセージを複数のメッセージキューに配信し、さらに複数のブローカーに配信することができます。

メッセージ送信のフォールトトレランスメカニズム:

メッセージを送信する側として、プロデューサーには 3 つのフォールト トレランス メカニズムがあります。

• ローカルキャッシュ: NameSever がクラッシュするのを防ぐために、NameSever から取得した情報をローカルにキャッシュします。
• 利用できないブローカー セット: プロデューサーにはブローカーのフォールト トレランス メカニズムがあり、これは sendLatencyFaultEnable スイッチを使用してオンにできます。 RocketMq は障害のあるブローカーの HashMap を維持し、特定のレイテンシ レベルのブローカーをこのマップに配置します。次にブローカーを選択するときには、利用できないブローカーは回避されます。
• 再試行: プロデューサーがメッセージを送信すると、再試行メカニズムが働き、デフォルトでは 3 回再試行されます。デッドレターキュー消費者の消費再試行が指定回数を超え、デッドレターキューに入ります

このようにして、トピックのメッセージを複数のメッセージキューに配信し、さらに複数のブローカーに配信することができます。

2. メッセージは誰に送信され、どこに保存されますか?

プロデューサーがNameSeverに接続する

プロデューサーは、ネームサーバーを介して指定されたトピックのブローカー ルーティング情報を取得し、トピックが持つメッセージ キュー、メッセージ キューが配置されているブローカー、ブローカーの IP とポートなどのデータのキャッシュをローカルに保存します。プロデューサーはマスター ブローカーにのみメッセージを送信し、スレーブはマスターとスレーブの同期を通じてデータを取得します。

では、Produce は NameSever にどのように接続するのでしょうか?

• 接続: 単一のプロデューサーがネームサーバーとの長い接続を維持し、トピック構成情報を定期的に照会します。ネームサーバーに障害が発生した場合、プロデューサーは利用可能な接続が確保され、自動的に再接続できるようになるまで、次のネームサーバーに自動的に接続します。
• ポーリング時間: デフォルトでは、プロデューサーはネームサーバーからすべてのトピックの最新のキュー ステータスを 30 秒ごとに取得します。つまり、ブローカーがダウンした場合、プロデューサーがそれを検出するまでに最大 30 秒かかります。この期間中、ブローカーに送信されたメッセージは失敗します。この時間は DefaultMQProducer の pollNameServerInteval パラメータによって決定され、手動で構成できます。
• ハートビート: ネームサーバーとのハートビートなし

プロデューサーがブローカーに接続

• 接続: プロデューサーは、トピックに関与するすべてのブローカーとの長期的な接続を維持します。
• ハートビート: デフォルトでは、プロデューサーは 30 秒ごとにすべてのブローカーにハートビートを送信します。ブローカーは 10 秒ごとにすべての存続している接続をスキャンします (この時間は変更できません)。接続が 2 分以内にハートビート データを送信しない場合 (現在の時刻と最終更新時刻の差が 2 分を超える場合、この時間は変更できません)、接続は閉じられます。

プロデューサーがブローカーに接続すると、メッセージはポーリングを通じてブローカーに送信され、元のメッセージを保存するブローカーの CommitLog に保存されます。キューに配信されたメッセージの位置情報を保存するための ConsumeQueue もあります。もちろん、メッセージ キューはディスクに保存されるため、メモリには影響しません。また、メッセージは定期的にクリーンアップされます。

では、消費されたメッセージはどこに行くのでしょうか?物理メッセージ ファイルはいつクリーンアップしますか?このデザインの利点は何ですか?

これらについては次の記事、ブローカーの記事で取り上げます。ブローカーという頭脳が RocketMQ がこのような高いスループットをサポートするのにどのように役立つかを徹底的に理解していただけます。

つまり、この質問は徹底的に研究する価値があります。面接中に、RocketMQ の使用方法だけでなく、そのストレージ原理やアドレス指定原理についても説明できれば、面接官はあなたに惚れ込むでしょう。この時点で、次の大きな課題は、重複したメッセージの処理方法、メッセージの順序の保証方法、分散システムでの分散トランザクションの保証方法など、さまざまな実用的な問題を解決することです。

面接官は、その場であなたにオファーを出し、「当社で働くといくらの給料がもらえると思いますか？」と尋ねます。

3. メッセージの種類

RocketMQ メッセージは、通常メッセージ、スケジュールおよび遅延メッセージ、シーケンシャル メッセージ、トランザクション メッセージの 4 つのタイプに大別できます。ここがポイントです！

4つのタイプを簡単に紹介

• 通常メッセージ: スケジュールされたメッセージや遅延メッセージ、シーケンス メッセージ、および機能付きのトランザクション メッセージとは区別される、RocketMQ バージョンのメッセージ キュー内の機能のないメッセージ。
• スケジュールされたメッセージと遅延メッセージ: メッセージ プロデューサーが指定されたメッセージの配信をスケジュール (遅延) できるようにします。最大 40 日間のサポートが可能です。
• 順次メッセージ: メッセージ コンシューマーが送信された順序でメッセージを消費できるようにします。
• トランザクション メッセージ: トランザクションの最終的な一貫性状態を実現するために、X または Open XA に似た分散トランザクション機能を実装します。

メッセージ キュー RocketMQ によって提供される 4 つのメッセージ タイプに対応するトピックを混在させることはできません。たとえば、作成された通常メッセージのトピックは通常メッセージの送受信にのみ使用でき、他の種類のメッセージの送受信には使用できません。同様に、トランザクション メッセージのトピックはトランザクション メッセージの送受信にのみ使用でき、他の種類のメッセージの送受信には使用できません。

一般的なメッセージ

通常メッセージ: メッセージ キュー RocketMQ 内の機能のないメッセージ。スケジュールされたメッセージや遅延メッセージ、シーケンシャル メッセージ、機能のあるトランザクション メッセージとは異なります。

通常のメッセージを送信する方法には、同期 Sync、非同期 Async、一方向 Oneway の 3 つがあります。

同期とは、メッセージを送信した後、次のメッセージを送信する前にサーバーが応答するのを待つ必要があることを意味します。非同期は、時間に敏感なビジネス シナリオに適しています。非同期では、サーバーの応答を待たずにメッセージを継続的に送信できます。一方向は非同期よりも時間がかかりませんが、通常はマイクロ秒レベルです。ただし、サーバーの応答を待たずに送信し、コールバック関数をトリガーしないため、信頼性は低下します。

同期送信

同期では、メッセージ送信者がメッセージを送信した後、サーバーからの応答を受信して​​から次のメッセージを送信します。

非同期送信

非同期送信とは、送信者がメッセージを送信し、サーバーが応答を返すのを待たずに次のメッセージを送信する通信方法を指します。

RocketMQ バージョンのメッセージ キューを非同期に送信するには、非同期送信コールバック インターフェイス (SendCallback) の実装が必要です。メッセージを送信した後、メッセージ送信者はサーバーの応答を待たずに 2 番目のメッセージを送信できます。送信者はコールバックインターフェースを介してサーバー応答を受信し、応答結果を処理する。

通常、時間に敏感なビジネスシナリオで使用されます

一方向送信

送信者はメッセージを送信する責任のみを負い、サーバーが応答を返すのを待たず、コールバック関数もトリガーされません。つまり、応答を待たずにリクエストを送信するだけです。この方法では、メッセージの送信に非常に短い時間（通常はマイクロ秒単位）しかかかりません。

ログ収集など、信頼性要件が高くないシナリオに適用可能

時間指定および遅延メッセージ

スケジュールされたメッセージと遅延メッセージ: メッセージプロデューサーが指定されたメッセージの配信をスケジュール (遅延) できるようにします。最大 40 日間のサポートがあります。

遅延メッセージは、メッセージが RocketMQ バージョンのメッセージ キューのサーバーに送信された後、一定時間後にクライアントに配信されて消費されるように指定するために使用されます (たとえば、3 秒後に消費されます)。これは、メッセージの生成と消費に時間枠の要件があるシナリオや、遅延キューと同様にメッセージによって遅延タスクがトリガーされるシナリオを解決するのに適しています。

スケジュールされたメッセージは、指定されたタイムスタンプの後にのみコンシューマーによって消費されます。これらは、メッセージの生成と消費に時間枠の要件があるシナリオや、スケジュールされたタスクをトリガーするためにメッセージが使用されるシナリオに適しています。

適用可能なシナリオ

一部のスケジュールされたタスクはメッセージによってトリガーされます。このとき、特定の時間にユーザーに送信されるリマインダー メッセージなど、スケジュールされたメッセージが役立ちます。タイムアウトや支払いの失敗により注文がクローズされるという典型的な電子商取引のシナリオなど、一部のメッセージの生成と消費の間には時間枠があります。このとき遅延メッセージが役立ち、期限内に支払いが完了しない場合は注文が閉じられます。

タイミングメッセージの精度には1秒～2秒の遅延誤差があります。

実際、スケジュールされたメッセージと遅延されたメッセージの使用時には、いくつかの違いがあります。使ったことのある人なら誰でも知っているはずです。スケジュールされたメッセージの場合、メッセージ送信時点後の特定の時点をメッセージ配信時点として明確に指定する必要があります。遅延メッセージには遅延時間の長さを設定する必要があります。長さは決まっていますが、時点は決まっていません。メッセージが送信される時点に関係します。メッセージは、現在の送信時点から一定時間遅延されてから配信されます。誰もがこれについて明確に理解しておくべきです。 Taobaoで注文すると、支払いに30分の猶予が与えられます。期限内にお支払いいただけない場合、注文は終了します。

連続メッセージ

順次メッセージ: メッセージコンシューマーが送信順にメッセージを送信できるようにします。

連続メッセージは 2 つのカテゴリに分けられます。

• グローバル順序: 指定されたトピックの場合、すべてのメッセージは厳密な先入れ先出し (FIFO) 順序で公開および消費されます。
• パーティション順序: 指定されたトピックの場合、すべてのメッセージはシャーディング キーに基づいてブロックにパーティション分割されます。同じパーティション内のメッセージは、厳密な FIFO 順序で公開および消費されます。シャーディング キーは、連続したメッセージ内の異なるパーティションを区別するために使用されるキー フィールドであり、通常のメッセージのキーとはまったく異なる概念です。

実はこれも定番の質問で、面接でもよく聞かれます。注文を確実にするにはどうすればいいですか?いずれにしても、ゆうゆが答えてくれるでしょう？

この問題に遭遇した場合、まずさまざまな状況でそれを説明する必要があります。つまり、グローバル順序とパーティション順序の 2 つの状況に分けられます。

1. グローバル順序は、パフォーマンス要件が高くなく、すべてのメッセージを厳密に先入れ先出しの順序で公開および消費する必要があるシナリオに適しています。私はこのような状況に遭遇したことがなく、通常はグローバル順序付けを使用しません。

2. パーティション順序は、高いパフォーマンス要件に適しています。シャーディング キーはパーティション フィールドとして使用され、データは先入れ先出しの順序に厳密に従ってブロック内で公開および消費されます。たとえば、ユーザーが登録すると、検証コードはユーザー ID をシャーディング キーとして使用するため、同じユーザーによって送信されたメッセージは公開された順序で消費されます。もう 1 つの例は、電子商取引における注文プロセスの問題です。

Alibaba グループの内部電子商取引システムはすべて、分割された順次メッセージを使用しており、これにより業務の順序が保証されるだけでなく、業務の高パフォーマンスも保証されます。なぜ私がこれを知っているのか聞かないでください。これはAlibaba Cloudの公式ウェブサイトに書かれています。

連続メッセージに関するよくある質問

グローバル シーケンシャル メッセージのパフォーマンスが平均的であるのはなぜですか?

グローバル シーケンシャル メッセージは FIFO メッセージ ブロッキング原則に厳密に従います。つまり、前のメッセージが正常に消費されない場合、次のメッセージはトピック キューに格納されます。グローバル シーケンシャル メッセージの TPS を向上させるには、インスタンス構成をアップグレードし、同時に、メッセージ クライアント アプリケーションでローカル ビジネス ロジックの処理に費やす時間を最小限に抑える必要があります。

シーケンシャルメッセージはどのようなメッセージ送信方法をサポートしていますか?クラスター消費とブロードキャスト消費をサポートしていますか?

シーケンシャル メッセージは信頼性の高い同期送信のみをサポートし、非同期送信はサポートしません。そうしないと、順序を厳密に保証できなくなります。シーケンシャル メッセージは現在、ブロードキャスト消費モードではなく、クラスター消費モードのみをサポートしています。

取引メッセージ

トランザクションメッセージ: 最終的な一貫性を実現するために、X または Open XA に類似した分散トランザクション機能を実装します。

RocketMQ バージョンのメッセージ キューは、X または Open XA に類似した分散トランザクション機能を提供します。 RocketMQ バージョンのメッセージ キューのトランザクション メッセージを通じて、分散トランザクションの最終的な一貫性を実現できます。

セミトランザクション メッセージ: 一時的に配信できないメッセージ。送信者はメッセージをメッセージ キュー RocketMQ サーバーに正常に送信しましたが、サーバーはプロデューサーからのメッセージの 2 次確認を受信して​​いません。この時点で、メッセージは「一時的に配信できません」と表示されます。この状態のメッセージはセミトランザクション メッセージです。

メッセージの再確認: ネットワークの中断、プロデューサー アプリケーションの再起動などの理由により、トランザクション メッセージの二次確認が失われます。 RocketMQ バージョンのメッセージ キュー サーバーは、スキャンによってメッセージが「セミトランザクション メッセージ」に長時間含まれていることを検出すると、メッセージの最終ステータス (コミットまたはロールバック) についてメッセージ プロデューサーに積極的に問い合わせる必要があります。この問い合わせプロセスはメッセージの再確認です。

トランザクション メッセージを送信する手順を見てみましょう。

1. 送信者は、セミトランザクション メッセージをサーバー ブローカーに送信します。サーバーはメッセージを保持し、メッセージが正常に送信されたことを確認するために ACK を返します。この時点で、メッセージはセミトランザクション メッセージです。

2. 送信者はローカルトランザクションのロジックの実行を開始する

3. 送信者は、ローカル トランザクションの実行結果に基づいて 2 番目の確認をサーバーに送信し、コミットするかロールバックするかを決定します。コミットを受信すると、サーバーはメッセージを配信可能としてマークし、コンシューマーに送信します。ロールバックを受信すると、サーバーはセミトランザクション メッセージを削除します。サーバーはそれを送信せず、消費者はそれを受信しません。

しかし、ネットワークが切断されたり、アプリケーションが再起動されたりすると、上記の手順の二次確認情報がサーバーに届かなくなります。どうすればいいでしょうか?

実際にはここにはチェックバック メカニズムが存在します。送信者がメッセージを送信した後、トランザクションをローカルで実行する必要があります。トランザクション実行プロセスが停止した場合、またはネットワークの問題によりトランザクション実行結果をサーバーに送信できない場合、サーバーはチェックバック メカニズムを実行して、セミトランザクション メッセージの最終的な送信ステータスを確認します。

要約する

RocketMQ メッセージ キューのコンシューマー クライアント オブジェクトとプロデューサー クライアント オブジェクトはスレッドセーフであり、複数のスレッド間で共有できます。複数のプロデューサー インスタンスとコンシューマー インスタンスを 1 台のサーバー (または複数のサーバー) にデプロイしたり、マルチスレッドを使用して同じプロデューサー インスタンスまたはコンシューマー インスタンスでメッセージを送受信したりして、メッセージの送受信の TPS を向上させることができます。スレッドごとに 1 つのクライアント インスタンスを作成することは避けてください。

さて、この記事の内容を振り返ってみましょう。

1. メッセージ送信の負荷分散とフォールトトレランスメカニズム

2. メッセージの送信プロセスと保存（これらはブローカーの CommitLog と ConsumerQueue に保存されるため、具体的な保存方法についてはブローカーのセクションで説明します）

3. メッセージの種類: 通常メッセージ (同期送信、非同期送信、一方向送信)、時間指定および遅延メッセージ、シーケンシャル メッセージ (グローバル シーケンスおよび部分シーケンス)、トランザクション メッセージ