[[323531]]

RocketMQは、ピュアJavaで開発された、分散型のキューベースのオープンソースメッセージミドルウェアです。以前はMetaQとして知られていたこのミドルウェアは、Alibabaによって開発されたキューベースのメッセージミドルウェアで、その後Apache Software Foundationによってオープンソース化され、Apacheのトップレベルプロジェクトとなりました。高性能、高信頼性、優れたリアルタイム性、そして分散アーキテクチャを特徴としています。

同時に、非同期通信分離、エンタープライズソリューション、金融支払い、電気通信、電子商取引、高速物流、広告とマーケティング、ソーシャルネットワーキング、インスタントメッセージング、モバイルアプリケーション、モバイルゲーム、ビデオ、モノのインターネット、車両のインターネットなど、多くの分野で広く使用されています。

次のような特徴があります。

• 厳密なメッセージ順序を保証できる
• 豊富なメッセージ取得モードを提供
• 非常に効率的な加入者レベルのスケーリング機能
• リアルタイムメッセージサブスクリプションメカニズム
• 数億件のメッセージを処理する能力

RocketMQのアーキテクチャと原理分析

RocketMQアーキテクチャ

ネームサーバー:

• 軽量なサービス検出とルーティングを提供します。ネームサーバーはブローカークラスターからの登録を受け付け、ブローカーがまだ存在するかどうかを確認するためのシグナリングメカニズムを提供します。
• 各ネームサーバーは、完全なルーティング情報 (ブローカー関連のトピックなどのメタ情報、コンシューマーがブローカー情報を見つける方法に関する情報をプロデューサーに提供) を記録し、対応する読み取りおよび書き込みサービスを提供します。

ブローカー (メッセージ サーバー): メッセージ ストレージ センター。プロデューサーからのメッセージを受信して​​保存します。コンシューマーはここからメッセージを取得します。

• 各ブローカー ノードは、すべてのネーム サーバー ノードとの永続的な接続とハートビートを維持し、定期的にトピック情報をネーム サーバーに登録します (基盤となる通信は Netty に基づいています)。
• ブローカーはメッセージの保存を担当し、トピックに基づく軽量キューをサポートします。1台のマシンで数万個のキューをサポートし、メッセージのプッシュ/プルモデルをサポートします。
• メッセージの秩序性を厳密に保証しながら、数億のメッセージを処理する能力を備えています。

プロデューサー：

• プロデューサーはメッセージを生成する役割を担い、ビジネス アプリケーション システムによって生成されたメッセージをメッセージ サーバーに送信します。
• プロデューサーは分散デプロイメントをサポートします。分散プロデューサーは、様々な負荷分散モードを用いてブローカークラスタにメッセージを送信します。送信プロセスは、高速な障害検出と低レイテンシをサポートします。
• メッセージを送信する方法は、同期、非同期、一方向の 3 つです。

消費者：

• コンシューマーはメッセージを消費する役割を担い、メッセージ サーバーから情報を取得してユーザー アプリケーションに入力します。
• また、「プッシュ アンド プル」モデルでの分散展開もサポートします。
• また、クラスタ利用とメッセージブロードキャストもサポートしており、リアルタイムメッセージサブスクリプションメカニズムを提供することで、ほとんどのコンシューマーのニーズを満たすことができます。

ブローカーサーバー

ブローカー サーバーは、メッセージの保存と送信、メッセージの取得、高可用性 (HA) などを担当します。ブローカー サーバーは、いくつかのメイン モジュールで構成されています。

リモート モジュール: クライアントからの要求を処理するブローカーのエントリ ポイント。

クライアント マネージャー: クライアント (プロデューサー/コンシューマー) を管理し、コンシューマー トピックのサブスクリプションを維持します。

ストア サービス (ストレージ サービス): データベースにメッセージを保存または取得するためのシンプルな API を提供します。

HA サービス (高可用性サービス): マスター ブローカーとスレーブ ブローカー間のデータ同期機能を提供します。

インデックス サービス: メッセージのインデックスを作成し、高速クエリ機能を提供します。

RocketMQの全体的なプロセス

1. ネームサーバーを起動します。起動後、ネームサーバーはポートをリッスンし、ブローカー、プロデューサー、コンシューマーからの接続を待機します。ルーティング制御センターとして機能します。

2. ブローカーが起動し、すべての Namesrv インスタンスとの永続的な接続を維持し、ハートビート パケットを定期的に送信します。

• ハートビート パケットには、現在のブローカー情報 (IP + ポートなど) が含まれており、すべてのトピック情報が保存されます。
• 登録が成功すると、Namesrv クラスターにはトピックとブローカー間のマッピング関係が確立されます。

3. メッセージを送受信する前に、トピックを作成します。トピックを作成する際は、トピックを保存するブローカーを指定する必要があります。また、メッセージの送信時にトピックを自動的に作成することもできます。

4. プロデューサーがメッセージを送信する

• 起動すると、まずクラスター内の Namesrv ノードの 1 つとの永続的な接続が確立され、送信中の現在のトピックが Namesrv から取得され、どのブローカーに存在するかが判断されます。
• 次に、対応するブローカーとの永続的な接続を確立し、ブローカーに直接メッセージを送信します。

5. 消費者の消費メッセージ

• Namesrv サーバーの 1 つとの永続的な接続を確立して、現在サブスクライブされているトピックが存在するブローカーを取得します。
• その後、ブローカーと直接接続チャネルが確立され、メッセージの消費が開始されます。 *RocketMQのメッセージドメインモデル

RocketMQ メッセージ

トピック: 第 1 レベルのメッセージ タイプを表し、サブスクリプションの最も細かい単位です (プロデューサーがメッセージを配信し、コンシューマーがメッセージ ID を取得します)。

• メッセージにはトピックが必要です
• グループは複数のトピックからのメッセージをサブスクライブできます。
• トピックは通常、トランザクション メッセージなどの研究の範囲または領域です。

タグ: メッセージの第2レベルのタイプを表します。同じトピックを使用しながらも異なるタグを使用することで、同じビジネスモジュール内の異なるタスクを表すメッセージとなる場合があります。例えば、トランザクションメッセージは、トランザクション作成メッセージ、トランザクション完了メッセージなどにさらに分類できます。

• コードをクリーンで一貫性のある状態に保つのに役立ちます
• RocketMQが提供するクエリシステムを簡素化する

メッセージ（メッセージ本体）：メッセージは伝達される情報です。メッセージにはトピックが必須で、オプションでタグとキーと値のペアが含まれます。

メッセージ キュー: すべてのメッセージ キューは永続的です。

• トピックは複数のキューを持つことができます
• キューの導入により、分散されたクラスター化されたメッセージ ストレージが可能になり、水平方向のスケーラビリティが実現します。

グループ:プロデューサーグループとコンシューマーグループに分かれており、同じ役割を持つグループがまとめられます。

• トランザクション後に元のプロデューサーがクラッシュした場合、ブローカーは同じプロデューサー グループ内の別のプロデューサー インスタンスに連絡して、トランザクションをコミットまたはロールバックできます。
• コンシューマー グループ内のコンシューマー インスタンスには、同一のトピック サブスクリプションが必要です。

RocketMQの機能

メッセージ モデル:

• クラスタリング: クラスター化された消費モードを使用する場合、MQ は、すべてのメッセージはクラスター内の 1 つのコンシューマーによってのみ処理される必要があると想定します。
• ブロードキャスト: ブロードキャスト消費モードを使用する場合、MQ は各メッセージをクラスター内のすべての登録済みクライアントにプッシュし、各マシンでメッセージが少なくとも 1 回消費されるようにします。

メッセージの順序

• DefaultMQPushConsumer を使用する場合、メッセージを順次使用するか、同時に使用するかを決定できます。

順次使用が指定されている場合、メッセージ使用の最大同時実行数は、コンシューマー グループがサブスクライブするメッセージ キューの数になります。

このモードでは、メッセージの順序は保証されなくなります。

• 同時: メッセージが同時に使用される場合、メッセージ使用の最大同時実行性は、各コンシューマー クライアントに指定されたスレッド プールによってのみ制限されます。
• 順序どおり: 順序どおりのメッセージ消費とは、プロデューサーが各メッセージキューにメッセージを送信した順序と同じ順序でメッセージが消費されることを意味します。(グローバルな順序を適用する必要がある状況に対処する必要がある場合は、トピックにメッセージキューが 1 つだけあることを確認してください。)

メッセージの種類

• 取引メッセージ
• 連続メッセージ
• 遅延メッセージ

RocketMQ スタンドアロンバージョンのインストール

1. ソースコードをダウンロードしてコンパイルします。

 # ダウンロード $
 > wget wget http://mirror. bit .edu.cn/apache/rocketmq/4.6.0/rocketmq- all -4.6.0-source- >
 # 解凍 $
 > rocketmq-all-4.7.0-source-release.zipを解凍します。
 > cd rocketmq- all -4.7.0/
 # コンパイル $
 > mvn -Prelease- all -DskipTests クリーンインストール -U
 > cd ディストリビューション/ターゲット/rocketmq-4.7.0/rocketmq-4.7.0

2. ネームサーバーを起動する

#ネームサーバーサービスを開始する
nohup sh bin/mqnamesrv &
 # 起動後、ログを確認してください。
 > tail -f ~/logs/rocketmqlogs/namesrv.log
ネームサーバーの起動に成功しました...

3. ブローカーを開始

conf ディレクトリでは、RocketMQ はさまざまなブローカーの構成ファイルを提供します。

 # ブローカーサービスを開始する
> nohup sh bin/mqbroker -n localhost:9876 &
 # 起動後、ログを確認してください。
 > tail -f ~/logs/rocketmqlogs/broker.log
ブローカー[%s、172.30.30.233:10911]の起動に成功しました...

そのうち、パラメータは次のとおりです。

• `-c` パラメータは、メインのブローカー構成を読み取るように構成します。
• RocketMQ Namesrv アドレスは -n パラメータを使用して設定されます。
• broker.conf: 単一マスター、非同期ディスクフラッシュ。
• 2m/: デュアルマスター、非同期ディスクフラッシュ。
• 2m-2s-async/: 2 つのマスターと 2 つのスレーブ、非同期レプリケーション、非同期ディスク フラッシュ。
• 2m-2s-sync/: マスター 2 つとスレーブ 2 つ、同期レプリケーション、非同期ディスク フラッシュ。
• dledger/: 少なくとも 3 つのノードを必要とする Dledger クラスター。

4. メッセージの送受信

メッセージを送受信する前に、クライアント（プロデューサー/コンシューマー）にネームサーバーのアドレスを通知する必要があります。RocketMQは、これを実現するためのいくつかの方法を提供しています。

簡単にするために、以下に示すように環境変数 NAMESRV_ADDR を使用します。

 #ネームサーバーのアドレスを設定する $
 > NAMESRV_ADDR=localhost:9876 をエクスポート
# プロダクションメッセージ$
 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
 SendResult [sendStatus=SEND_OK, msgId= ...
 # 消費者ニュース $
 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
 ConsumeMessageThread_%d 新しいメッセージを受信: [MessageExt...

• コードでは、producer.setNamesrvAddr("ip:port") を設定します。
• Javaプロパティ設定: rocketmq.namesrv.addr
• 環境変数の設定: NAMESRV_ADDR
• HTTPエンドポイント