メッセージキュー技術とRabbitMQの基本

I. メッセージミドルウェアの特徴

1. 非同期処理モデル

メッセージ送信者は応答を待つことなくメッセージを送信できます。送信者はメッセージを仮想のチャネルまたはキューに送信し、受信者はそのチャネルを購読または監視します。メッセージは最終的に1つまたは複数のメッセージ受信者に転送され、これらの受信者はメッセージ送信者に同期応答をする必要はありません。プロセス全体は非同期であり、ユーザー登録などに例えられます。

2. アプリケーション間の疎結合

アプリケーション間の呼び出し関係は疎結合であり、送信者と受信者は互いを理解する必要がありません。メッセージの確認のみが必要で、送信者と受信者は同時にオンラインである必要はありません。例えば、オンライン取引システムではデータ一貫性を保証するために、支払いシステム処理完了後に支払い結果をメッセージミドルウェアに格納し、注文システムに注文支払い状態の変更を通知します。これにより、2つのシステムはメッセージミドルウェアによってデカップリングされます。

II. メッセージミドルウェアの配送モデル

1. ポイントツーポイントモデル(PTP)

ポイントツーポイントモデルは、メッセージプロデューサーとコンシューマー間の1対1の通信に使用されます。メッセージプロデューサーは特定のコンシューマーにメッセージを送信し、これはサービス内のキューに対応します。メッセージがコンシューマーに配送される前に、キューに保存されます。

1.1 モデルの特性:

  • 各メッセージには1つのコンシューマーのみが存在します
  • 送信者とコンシューマーには時間的依存性がなく、受信者がメッセージの受信と処理を確認します

2. パブリッシュ/サブスクライブモデル(PUB/SUB)

パブリッシュ/サブスクライブモデルは、特定のメッセージトピックにメッセージを生成することをサポートします。0または複数のサブスクライバーが特定のメッセージトピックからのメッセージに興味を持つ可能性があります。このモデルでは、パブリッシャーとサブスクライバーは互いを知りません。このモデルは匿名公告のようなものです。

2.1 パブリッシュ/サブスクライブモデルの特性

  • 各メッセージには複数のサブスクライバーが存在します
  • クライアントは購読後にのみメッセージを受信できます
  • パブリッシャーとサブスクライバー間には時間的依存性が存在します

2.2 永続的サブスクリプションと非永続的サブスクリプション

永続的サブスクリプションが故障した場合、再起動後にメッセージが再送信されます。非永続的サブスクリプションの場合、メッセージは失われます。

III. 使用例

1. 非同期処理 - ユーザー登録

ユーザー登録が成功すると、メールまたはSMSが送信されます。ネットワークの揺れやサービスの停止が発生した場合、メッセージのタイムアウトメカニズムまたはリトライメカニズムが必要になることがあります。

2. ログ分析

PV計算やユーザー行動分析などで使用され、サブスクライブモデルが適用されます。

3. データ複製

データをソースから複数の宛先に複製し、データの一貫性、順序、または因果シーケンスを維持します。クロスデータセンターデータ転送、検索、オフラインデータ計算などに使用されます。

4. 遅延メッセージ送信と一時保存

メッセージミドルウェアを信頼できる一時保存場所として使用し、定期的にメッセージを配信します。例えば、ユーザーの秒殺イベントのシミュレーションや負荷テストなどに使用されます。

5. メッセージブロードキャスト

キャッシュデータの同期更新、アプリケーションからのプッシュデータなどに使用されます。例えば、ローカルキャッシュの更新などに使用されます。

IV. メッセージミドルウェアの分類

メッセージミドルウェアは主にpushとpullの2つのモデルに分けられます。

1. pushモデル

メッセージプロデューサーがメッセージをメッセージングサービスに送信し、サービスがメッセージをコンシューマーにプッシュします。

2. pullモデル

コンシューマーがメッセージサービスからメッセージを要求して受信します。

一般的なシナリオでは、コンシューマーがプルする方が制御が容易です。プロデューサーがコンシューマーにプッシュする場合、負荷分散、バックエンドサブスクライバー情報収集のクラスターなど、より多くの処理が必要です。一方、プルモデルは制御範囲が容易ですが、一度に大量のデータをプルする可能性があり、データストレージデバイスと帯域幅の考慮も必要です。リアルタイム情報送信のようなシナリオではプッシュモデルがより適しています。

V. MetaQの紹介

MetaQはJava言語で記述され、複数のプラットフォームで展開できます。クライアントはC++とJavaをサポートし、単一サーバーで1万以上のメッセージキューをサポートできます。実際のテストでは4〜6千程度ですが、水平方向にスケールアウトできます。キューは永続化され、キューの長さはディスクサイズによって制限されます。

MetaQの特徴:

  • プロデューサー、サーバー、コンシューマーはすべて分散可能
  • メッセージは通常順次書き込みで保存され、性能が非常に高い
  • メッセージの順序をサポート
  • クライアントプル、ランダム読み取り、バッチデータ取得、データ移行、拡張をサポート
  • 同名ユーザーをサポート
  • 消費状態はクライアント側に保持される

MetaQ用語

  • topic: メッセージのトピック
  • offset: メッセージのオフセット値
  • broker: MetaQのサーバー
  • partition(パーティション): 同じトピック内の複数のパーティション

MetaQの主な設定

  • zk.zkEnable: ZooKeeperへの登録設定
  • zk.zkConnect: ZooKeeperサーバーリスト
  • brokerId: サーバーID（0-1024）
  • serverPort: サービスポート
  • dataPath: データストレージパス
  • deletePolicy: データ削除ポリシー
  • flushTxLogAtCommit: トランザクションログの同期設定

MetaQクラスタ

すべてのbrokerがZooKeeperに登録され、クライアントはZooKeeperに接続して利用可能なbrokerリストを取得し、1つのbrokerを選択してメッセージを送信します。

VI. RabbitMQ

RabbitMQはAMQPプロトコル標準に基づいた、完全で再利用可能なエンタープライズメッセージシステムです。Mozilla Public Licenseオープンソースプロトコルに従い、Erlangで実装されたインダストリアルグレードのメッセージキューサーバーです。

AMQP(Advanced Message Queuing Protocol)は、非同期メッセージ伝送に使用されるアプリケーション層プロトコル仕様です。JMSのようなAPIではなく、ワイヤレイヤープロトコルとして機能します。

RabbitMQ全体アーキテクチャ

RabbitMQの核心コンポーネントはexchangeとqueueです。

RabbitMQ用語

  • Server(broker): クライアント接続を受け入れ、AMQPメッセージキューとルーティング機能を実現するプロセス
  • Virtual Host: 仮想的な概念で、権限制御グループに似ています
  • exchange: プロデューサーからのメッセージを受け取り、bindingルールに基づいてメッセージをサーバーのキューにルーティングします
  • message queue: 未消費のメッセージを保存するキュー
  • message: headerとbodyで構成されます
  • binding key: 特定のexchangeと特定のqueueを結びつけるキー

Exchangeの種類

1. Direct Exchange

ルーティングキーを直接処理します。キューをexchangeにバインドする必要があり、メッセージは特定のルーティングキーと完全に一致する必要があります。例えば、キューがルーティングキー"dog"でexchangeにバインドされている場合、"dog"とマークされたメッセージのみが転送され、"dog.puppy"や"dog.guard"は転送されません。

2. Fanout Exchange

ルーティングキーをブロードキャストします。キューをexchangeにバインドするだけで、exchangeに送信されたメッセージはそのexchangeにバインドされたすべてのキューに転送されます。サブネットブロードキャストに似ており、サブネット内のすべてのホストがメッセージのコピーを受け取ります。fanout exchangeはメッセージ転送が最も高速です。

3. Topic Exchange

メッセージのルーティングキーとバインディングキーのパターンマッチングに基づいてメッセージをルーティングします。このルータータイプは、经典的なパブリッシュ/サブスクライブメッセージ伝送モデルをサポートするために使用できます。トピック名前空間をメッセージアドレス指定パターンとして使用し、部分または完全にトピックパターンに一致する複数のコンシューマーにメッセージを伝達します。

バインディングキーは0個以上のタグで構成され、各タグは"."文字で区切られます。バインディングキーはこの形式で明示的に指定する必要があり、ワイルドカード「*」（単一の単語を一致）と「#」（ゼロまたは複数の単語を一致）をサポートします。例えば、バインディングキー"*.stock.#"はルーティングキー"usd.stock"と"eur.stock.db"に一致しますが、"stock.nasdaq"には一致しません。

設定

設定ファイルは/etc/rabbitmq/rabbitmq-env.confに保存されます。主要なパラメータには以下のものがあります：

  • tcp_listerners: RabbitMQの監听ポートを設定（デフォルト5672）
  • disk_free_limit: ディスク低水準線。指定値以下の場合データ受信を停止
  • vm_memory_high_watermark: メモリ低水準線。指定値を超えるとフロー制御メカニズムが有効（デフォルト0.4）

インストール

yum install epel-release
yum install rabbitmq-server -y
rabbitmq-plugins enable rabbitmq_management
systemctl restart rabbitmq-server

仮想ホストの管理

仮想ホストの作成: rabbitmqctl add_vhost vhostname
仮想ホストの表示: rabbitmqctl list_vhosts
仮想ホストの削除: rabbitmqctl delete_vhost vhostname

ユーザーと権限の管理

ユーザー作成: rabbitmqctl add_user username password
パスワード変更: rabbitmqctl change_password username newpassword
権限設定: rabbitmqctl set_permissions -p vhost username ".*" ".*" ".*"
キューの表示: rabbitmqctl list_queues -p vhost

タグ: メッセージキュー RabbitMQ AMQP ミドルウェア 非同期処理

5月23日 06:24 投稿

ホットタグ

©2026 異端開発室