Skip to content

MS_RabbitMQ

Jump to bottom
nishi_74322014 edited this page Sep 1, 2026 · 1 revision

RabbitMQ

概要

  • AMQP ブローカーの OSS
  • MQTT プロトコルサポートのプラグインもある。

補足（RabbitMQ の位置付け）: 一言で言えば
**「自分で立てられる汎用メッセージ ブローカー」**である。
本 Wiki の他のメッセージング技術との関係は次のとおり。

RabbitMQ Azure Service Bus Mosquitto
主プロトコル AMQP 0-9-1（＋ MQTT/STOMP プラグイン） AMQP 1.0 MQTT
運用 自分で構築・運用 マネージド 自分で構築
用途 汎用（業務連携、非同期処理） 業務連携 IoT デバイス
費用 サーバ代のみ 従量課金 サーバ代のみ
特徴 Exchange による柔軟なルーティング 企業向け機能が充実 軽量

なお、AMQP 0-9-1 と AMQP 1.0 は別物である
（名前は同じだが互換性が無い）。

AMQP 0-9-1 … RabbitMQ が実装。Exchange / Queue / Binding という
              ブローカー モデルをプロトコルが規定する
      ↓ 大きく再設計
AMQP 1.0   … ISO/IEC 19464 の国際標準。
              【ブローカーの内部モデルは規定しない】（後述）

本文の「ブローカーモデル」の節で
**「AMQP 1.0 で削除されたので、コレは、純粋に、RabbitMQ の話」**と
注記されているのは、この経緯による。

詳細

Windows

インストール

補足（なぜ Erlang が必要なのか）: RabbitMQ は
Erlang/OTP で実装されているため、その実行環境が要る。

これは単なる実装の都合ではなく、設計上の必然である。

Erlang の特性 RabbitMQ での価値
軽量プロセスを大量に扱える 接続・キューごとにプロセスを割り当てられる
プロセスが独立（メモリを共有しない） 1 つが落ちても他に波及しない
スーパーバイザによる自動再起動 障害からの自己回復
分散が言語機能 クラスタリングが自然に実装できる

つまり、**「大量の接続を捌き、一部が壊れても全体は止まらない」**という
メッセージ ブローカーの要件に Erlang が適合している。

実務上の注意点として、

  • Erlang と RabbitMQ のバージョンには対応関係がある
    （公式の互換性一覧を確認する）、
  • Erlang を先にインストールする必要がある、
  • Windows では ERLANG_HOME が必要になる場合がある
    （本文では「不要だった」とあるが、環境による）

という点が挙げられる。

なお、現在は Docker で動かすのが最も手軽で、
Erlang のバージョン管理から解放される。

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

-management タグを使えば後述の管理画面も有効化済みで起動する。

ブローカー起動

  • サービスとしてインストールされる。
  • 起動：メニューから［RabbitMQ Service - start］を選択
  • 停止：メニューから［RabbitMQ Service - stop］を選択

送受信

その他の環境

Linux

Raspberry Pi

Docker on Windows

Docker on Raspberry Pi

移行メモ: 上記 4 項目は原典でも見出しのみで、本文が存在しない。

SaaS / PaaS

  • Azure：Azure Service Bus（ブリッジ機能）
  • Azure：Amazon MQ for RabbitMQ
  • GCP：Cloud Pub/Sub？

移行メモ（正誤）: 2 行目の「Azure：Amazon MQ for RabbitMQ」は、
Amazon MQ は AWS のサービスであるため
「AWS：Amazon MQ for RabbitMQ」の誤りと判断される。
原典どおりに記載したうえで、ここに訂正を記す。

補足（マネージドで RabbitMQ を使う選択肢）: 現在の主な選択肢は次のとおり。

提供 サービス 内容
AWS Amazon MQ for RabbitMQ RabbitMQ そのものをマネージドで提供
Azure － RabbitMQ のマネージド提供は無い（VM / AKS で自前構築、または Service Bus に移行）
GCP － 同上（Cloud Pub/Sub は別物）
サードパーティ CloudAMQP 各クラウド上で RabbitMQ をマネージド提供

Azure で RabbitMQ を使いたい場合、

方針 内容
Service Bus に移行する AMQP 1.0 なので、クライアント側の書き換えが要る
AKS / VM で自前運用 RabbitMQ のまま。運用は自分持ち
Container Apps 手軽に載せられるが、永続化と可用性の設計が要る

という判断になる。
なお、Cloud Pub/Sub は Pub/Sub 専用であり、
RabbitMQ の Exchange のような柔軟なルーティングは持たない。
「？」が付いているとおり、直接の対応物ではない。

管理ツール

管理画面

  • 設定

    • メニューから［RabbitMQ Command Prompt (sbin dir)］を選択

    • 以下の 2 つのコマンドを実行する。

      rabbitmq-plugins enable rabbitmq_management

      rabbitmq-plugins enable rabbitmq_tracing

  • アクセス

  • 参考

補足（管理画面は本番でそのまま公開しない）: 管理画面は
キューの中身の閲覧・削除、ユーザ管理までできる強力な機能である。

注意 内容
HTTP（平文）である 本文のとおり既定は HTTP。認証情報が平文で流れる
guest/guest 既定のまま。削除するか、パスワードを変更する
15672 番の公開 インターネットに晒さない

対処としては、

  • management.ssl.port で HTTPS 化する、
  • リバース プロキシの背後に置き、認証を前段で行う、
  • ファイアウォールで管理端末からのみ許可する

といった措置を取る。
クラウド利用時の注意事項の
「インバウンドを不用意に開けない」がそのまま当てはまる。

なお、rabbitmq_tracing プラグインは
流れるメッセージの中身を記録する機能で、
開発・調査には有用だが性能に影響し、機密情報も記録されるため、
本番では無効にするのが原則である。

rabbitmqctl

  • 標準の管理ツールで vhost 作成・ユーザ作成・権限付与などの管理作業を行う。

  • Windows の場合
    以下から使用する。

    • 格納場所は以下のディレクトリ。

      ...RabbitMQ Server\rabbitmq_server-x.x.x\sbin

    • .bat ファイルとして提供されている。

      rabbitmqctl.bat

    • メニューから［RabbitMQ Command Prompt (sbin dir)］を選択し、.bat は省略して実行可能。

      rabbitmqctl(.bat) status

  • 参考

移行メモ（体裁）: 原典の「.batは省略可能して実行可能」は
「.bat は省略して実行可能」の誤りであるため修正した。

rabbitmq-plugins

前述の rabbitmq-plugins でプラグインを制御できる。

rabbitmqadmin

  • 管理画面に付属するツール（REST API のクライアント）で、
    Exchange や Queue の作成、Message の送受信などを行う。

  • python が必要で、rabbitmqadmin ファイルを管理画面からダウンロードして、以下のように使用する。

    • Linux の場合
      rabbitmqadmin ファイルを /usr/local/bin にコピーし、

      $ rabbitmqadmin ...サブ・コマンド...

    • Windows の場合
      rabbitmqadmin ファイルをコピーし（sbin dir が良いかと）、

      >python.exe rabbitmqadmin ...サブ・コマンド...

  • Message の送受信

    • 受信

      • 例

        rabbitmqadmin -u admin -p admin -V my-vhost get queue=my-q ackmode=ack_requeue_true

      • 初期値

        rabbitmqadmin get queue=my-q ackmode=ack_requeue_true

    • 送信

      • 例

        rabbitmqadmin -u admin -p admin -V my-vhost publish routing_key=my-key exchange=my-exchange payload=test

      • 初期値

        rabbitmqadmin publish routing_key=my-key exchange=my-exchange payload=test

  • 参考

移行メモ（体裁）: 原典の「Linux ルの場合」は
「Linux の場合」の誤りであるため修正した。

補足（ackmode の指定が重要）: 受信の例にある
ackmode=ack_requeue_true は、
**「取り出すが、キューには戻す」**という指定である。

値 動作
ack_requeue_true 中身を見るが、消さない（調査向け）
ack_requeue_false 取り出して消す（本当に消費する）
reject_requeue_true 拒否して戻す

調査目的でキューを覗く際に ack_requeue_false を使うと、
本番のメッセージを消してしまう。
本文が既定値としても ack_requeue_true を挙げているのは
妥当な選択である。

...

参考

設定ファイル

新・旧の設定ファイル

  • 場所
    インストール後、設定ファイルは存在しないので、必要なら、作成する必要がある。

    • Windows：

      %APPDATA%\RabbitMQ\rabbitmq.conf
C:\Users\xxxxx\AppData\Roaming\RabbitMQ\rabbitmq.conf

    • Linux：

      /etc/rabbitmq/rabbitmq.config

  • 新ファイル

    • rabbitmq.conf
      • 新式の書き方。
      • sysctl 方式で記述。
    • advanced.config
      • 旧式の書き方。
      • 新スタイルの設定形式では表現できない限られた設定項目。

  • 旧ファイル

    • rabbitmq.config
      • 旧式の書き方。
      • Erlang の標準設定ファイル書式で記述。

  • 参考

補足（3 つのファイルの関係）: 紛らわしいので整理する。

ファイル 書式 位置付け
rabbitmq.conf key = value（sysctl 風） 現行の標準。ほとんどの設定はここ
advanced.config Erlang の項 rabbitmq.conf で書けないものだけ
rabbitmq.config Erlang の項 旧形式（3.7 以前）。現在は非推奨

両方が存在する場合は両方が読まれる（advanced.config が後勝ち）。
後掲のチュートリアルで advanced.config を使っているのは、
auth_mechanisms のようなリスト構造の設定を扱っているためである。

なお、Windows では %APPDATA% 配下という点が特徴的で、
サービスとして動くアカウントの %APPDATA% を見るため、
ログインしているユーザのフォルダとは異なる場合がある。
管理画面の「Overview → Config file」で
実際に読まれているパスを確認するのが確実である。

インポート・エクスポート

  • エクスポート

    rabbitmqadmin export c:\export.json

  • インポート
    rabbitmq.conf に以下を追記する。

    management.load_definitions = c:\import.json

補足（この機能が構成管理の要）: エクスポートされる JSON には
ユーザ、権限、vhost、Exchange、Queue、Binding、ポリシーが
すべて含まれる。

これにより、

用途 内容
環境の複製 開発環境の構成を検証・本番へそのまま移す
構成の版管理 JSON を Git に入れる（クラウド・インフラ自動化）
障害復旧 再構築時に一発で戻せる
コンテナ化 起動時に定義を読み込ませる（イミュータブルな運用）

ということが可能になる。

ただし、エクスポートにはパスワード ハッシュが含まれるため、
この JSON は機密情報として扱う必要がある。

手作業で管理画面から設定する運用は
再現性が無く、環境差異の温床になる。
Azureによる基盤開発法の
「膨大なパラメタ・シートを作成しない」という指針と同じく、
定義ファイルそのものを構成の正とするのが望ましい。

管理ツール

前述の管理ツールを使用して設定を行う。

認証

バーチャルホスト、キュー、エクスチェンジなどのリソースへのアクセスを承認する

パスワード

  • デフォルト・ユーザー
    • ユーザー名「guest」パスワード「guest」。
    • バーチャルホスト「/」へのフルアクセスを許可される。
    • 規定ではホスト・ローカル接続にしか使用できない。
      loopback_users の設定を none にすることで、
      リモートホストからの接続を許可することが可能。
  • ユーザの管理

    • CLI

      • 追加

        rabbitmqctl add_user 'username' 'password'

      • 更新

        rabbitmqctl change_password

      • 削除

        rabbitmqctl delete_user 'username'

      • 一覧

        rabbitmqctl list_users

      • アクセス許可設定
        Exchange以下（Queueなど）のアクセス許可を設定できる。
        （Topic の認可というのもあるらしい（既定では無効になっている））
        ・追加

        # First ".*" for configure permission on every entity
# Second ".*" for write permission on every entity
# Third ".*" for read permission on every entity
rabbitmqctl set_permissions -p "custom-vhost" "username" ".*" ".*" ".*"

        ・削除

        rabbitmqctl clear_permissions -p "custom-vhost" "username"

        ・一括

        for v in $(rabbitmqctl list_vhosts --silent); do rabbitmqctl set_permissions -p $v "a-user" ".*" ".*" ".*"; done

    • その他

      • ノードブート時の定義のインポート
      • 別の認証バックエンド（LDAP、HTTP、AMQP など）
  • 参考

補足（loopback_users の設定は慎重に）: 本文が触れている
**「loopback_users の設定を none にする」**は、
guest ユーザをリモートから使えるようにするという意味である。

【既定】 guest/guest はローカルホストからのみ
           ↑ 安全側の設計

【loopback_users = none】
         guest/guest でどこからでも接続できる
           ↑ 【誰でも知っている資格情報】が外部から使える

検証環境で一時的に緩めることはあっても、
本番でこの設定を使ってはならない。
正しい手順は、

  1. 専用のユーザを作成して権限を付与する、
  2. guest ユーザを削除する（rabbitmqctl delete_user guest）

である。

なお、3 つの ".*" は
configure / write / read の権限をそれぞれ正規表現で指定するもので、
".*" はすべてのリソースに対して許可を意味する。

権限 対象となる操作
configure Queue / Exchange の作成・削除
write メッセージの発行（publish）、Binding
read メッセージの取得（consume）

実務では、発行するだけのアプリ、消費するだけのアプリのように
最小権限で絞るのが望ましい。
".*" ".*" ".*" は「全権」であり、
Azureのアクセス制御と権限で言えば
「所有者」を配っているのと同じである。

証明書

補足（クライアント証明書認証の要点）: 「ユーザー名は CN と一致する」
「パスワードはすべて無視される」という 2 行が、
この方式の本質を表している。

【パスワード認証（PLAIN）】
  接続時に ユーザー名 + パスワード を送る
     ↓ パスワードが漏れると成りすまされる

【クライアント証明書認証（EXTERNAL）】
  TLS ハンドシェイクでクライアント証明書を提示
     ↓ その証明書の CN が「ユーザー名」として扱われる
  → パスワードという概念が無い

つまり、RabbitMQ 側に「CN と同じ名前のユーザ」を
作っておく必要がある（後掲のチュートリアルで
「CN と一致したユーザーを作成する」とあるのはこのため）。

利点 注意
パスワードを配布・管理しなくてよい 証明書の配布と失効管理が必要になる
秘密鍵は端末から出ない 有効期限切れで一斉に接続不能になる
経路も暗号化される 証明書ストアの管理（証明書）

IoT デバイスのように台数が多く、パスワード運用が現実的でない
場合に有効な方式である
（IoT関連の通信プロトコル）。

ブローカーモデル

AMQP 1.0 で削除されたので、コレは、純粋に、RabbitMQ の話。

Producer

送信する人

Consumer

受信する人

Message

送受信対象のメッセージ

Queue

キュー

Exchange

Queue への配送ルール

  • direct
    routing key と binding key が完全一致した Message だけ Queue に配送。
  • fanout
    binding key を無視して全ての Queue に Message を配送（ブロードキャスト）
  • topic
    routing key と binding key で、部分一致した Message を Queue に配送。

移行メモ（体裁）: 原典の「完全したMessage」は
「完全一致した Message」の脱字であるため補った。

補足（Exchange が RabbitMQ の中核）: 「Producer が Queue に直接入れる」
のではなく、必ず Exchange を経由するという構造が
RabbitMQ の特徴である。

Producer → 【Exchange】 --binding--> Queue1 → Consumer A
                 │       --binding--> Queue2 → Consumer B
                 ↑
           ここで「どの Queue に入れるか」を決める

この間接化により、Producer は宛先を知らなくてよい。
Queue や Consumer を後から増やしても、
Binding を足すだけで Producer は変更不要になる（疎結合）。

3 種類（＋ headers）の使い分けは次のとおり。

種類 判定 用途
direct 完全一致 キューへの振り分け
fanout 判定しない（全部に配る） 同報（Azure Event Gridのファンアウトに相当）
topic ワイルドカード一致（*＝1 語、#＝複数語） 階層的な振り分け
headers ヘッダーの値 特殊

topic が最も柔軟で、例えば routing key が log.error.db のとき、
binding が log.# や log.error.* なら一致し、
log.info.* なら一致しない。
1 つのメッセージを複数の観点で振り分けられる。

なお、Azure Service Busには Exchange に相当する
概念が無く、トピックのサブスクリプションにフィルタを書く形になる。
移行の際はこの構造の違いが設計変更を伴う。

Vhosts (Virtual Hosts)

  • 固有の名前を持つ個々のコンテナ（論理的グループ）
  • コンテナ中には、以下リソースがある。
    • Connection
    • Exchange
    • Queue
    • Binding
    • ユーザー権限
    • およびその他のシステムリソース

補足: vhost は
1 台の RabbitMQ を論理的に分割する仕組みである。

【1 つの RabbitMQ サーバ】
  ├─ vhost "/prod"  … 本番用の Exchange / Queue / 権限
  ├─ vhost "/stg"   … 検証用（名前が同じ Queue を作れる）
  └─ vhost "/team-a" … 部門用
効果 内容
名前空間の分離 同名の Queue を別 vhost に作れる
権限の分離 ユーザに vhost 単位で権限を与える
設定の分離 ポリシーも vhost 単位

ただし、リソース（メモリ・CPU・ディスク）は共有である点に注意する。
1 つの vhost で大量にメッセージが滞留すると、全体が止まる
（メモリ アラームで全接続がブロックされる）。
本番と検証を同じサーバの別 vhost に置くのは
この理由で推奨されない。

送受信（C#）

RabbitMQ.Client

様々な言語向けにリリースされている。

  • 以下のコードは.NET Coreでも大方動く。
    • 一箇所、変更した。

      • ea.Body

        ↓ ↓ ↓

      • ea.Body.ToArray()

    • Qiita のサンプルをベースに弄った。
using Newtonsoft.Json;
using RabbitMQ;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Net.Security;
using System.Security.Cryptography.X509Certificates;

namespace ConsoleApp1
{
    class Program
    {
        static void Main(string[] args)
        {
            // ホスト名
            var hostname = "localhost";

            // Taskキャンセルトークン
            var tokenSource = new CancellationTokenSource();

            Console.WriteLine($"start .Net RabbitMQ Example. Ctl+C to exit");

            X509Certificate2 cli = new X509Certificate2(@"C:\certs\client.pfx", "xxxxx");

            // ファクトリ生成
            var factory = new ConnectionFactory()
            {
                HostName = hostname,
                //UserName = "hogecli",
                //Password = "hogecli",
                AuthMechanisms = new IAuthMechanismFactory[] { new ExternalMechanismFactory() },
                Ssl = new SslOption
                {
                    Enabled = true,
                    ServerName = hostname,
                    AcceptablePolicyErrors = SslPolicyErrors.RemoteCertificateNameMismatch
                                             | SslPolicyErrors.RemoteCertificateChainErrors,
                    Certs = new X509CertificateCollection(new X509Certificate[] { cli })
                }
            };

            // パブリッシャータスク
            var pTask = Task.Run(() => new Action<ConnectionFactory, CancellationToken>(async (f, cancel) => {
                // コネクション＆チャンネル生成
                using (var conn = f.CreateConnection())
                using (var channel = conn.CreateModel())
                {
                    // Exchange生成
                    channel.ExchangeDeclare("test", "fanout", false, true);

                    while (true)
                    {
                        // キャンセル待ち
                        if (cancel.IsCancellationRequested)
                        {
                            break;
                        }

                        var msg = new SendMessage()
                        {
                            Message = "Hello",
                            Timestamp = DateTime.UtcNow.ToBinary()
                        };
                        var body = JsonConvert.SerializeObject(msg);

                        // Publish!!
                        try
                        {
                            channel.BasicPublish("test", "", null, Encoding.UTF8.GetBytes(body));
                            Console.WriteLine($"success send. message: {msg.Message}, timestamp: {msg.Timestamp}");
                        }
                        catch (Exception ex)
                        {
                            Console.WriteLine($"failer send. reason: {ex.Message}");
                        }

                        await Task.Delay(10000);
                    }
                }
            })(factory, tokenSource.Token), tokenSource.Token);

            // コンシューマータスク
            var cTask = Task.Run(() => new Action<ConnectionFactory, CancellationToken>((f, cancel) => {
                // コネクション＆チャンネル生成
                using (var conn = f.CreateConnection())
                using (var channel = conn.CreateModel())
                {
                    // Exchange生成
                    channel.ExchangeDeclare("test", "fanout", false, true);

                    // Queue生成
                    var queueName = channel.QueueDeclare().QueueName;

                    // Bind Queue
                    channel.QueueBind(queueName, "test", "");

                    // コンシューマー生成
                    var consumer = new EventingBasicConsumer(channel);

                    // 受信イベント定義
                    consumer.Received += (_, ea) =>
                    {
                        var msg = JsonConvert.DeserializeObject<ConsumedMessage>(Encoding.UTF8.GetString(ea.Body.ToArray()));
                        Console.WriteLine($"success consumed. message: {msg.Message}, timestamp: {msg.Timestamp}");
                    };

                    // コンシューマー登録
                    channel.BasicConsume(queueName, true, consumer);

                    while (true)
                    {
                        // キャンセル待ち
                        if (cancel.IsCancellationRequested)
                        {
                            break;
                        }
                    }
                }
            })(factory, tokenSource.Token), tokenSource.Token);

            // Ctl+C待機
            Console.CancelKeyPress += (_, e) =>
            {
                e.Cancel = true;
                tokenSource.Cancel(); // Taskキャンセル
            };

            Task.WaitAll(pTask, cTask);

            Console.WriteLine("stop .Net RabbitMQ Example. press any key to close.");
            Console.ReadKey();
        }
    }

    public class SendMessage
    {
        public string Message { get; set; }
        public long Timestamp { get; set; }
    }

    public class ConsumedMessage
    {
        public string Message { get; set; }
        public long Timestamp { get; set; }
    }
}

補足（ea.Body → ea.Body.ToArray() の変更が意味するもの）: 本文が
「一箇所、変更した」と述べているこの差分は、
.NET の性能改善の流れを反映した重要な変更である。

【旧】 ea.Body            … byte[]（配列。コピーが発生する）
         ↓ RabbitMQ.Client 6.0 で変更
【新】 ea.Body            … ReadOnlyMemory<byte>（コピーしない）
       ea.Body.ToArray()  … byte[] が要る場合は明示的に変換

**ReadOnlyMemory<byte> は「バッファへの参照」**であり、
配列を新たに確保しない。
これは JSONのparseを色々試してみた。で述べた
System.Text.Json が Span<T> を使うのと同じ動機である。

ただし、重大な注意点がある。
ハンドラの外に ea.Body を持ち出すと、内容が壊れる
（別のメッセージで上書きされる）。
非同期処理に渡す場合は必ず ToArray() でコピーする。

なお、RabbitMQ.Client 7.0 以降では API が大きく変わり、
完全に非同期（IChannel、CreateChannelAsync、
AsyncEventingBasicConsumer）になっている。
上のコードは 6.x 系までのものである。

補足（このサンプルの注意点）: 動作確認用のコードとしては明快だが、
本番でそのまま使ってはいけない点がいくつかある。

箇所 問題 対処
BasicConsume(queueName, true, ...) 第 2 引数が autoAck: true。受信した瞬間に ACK されるため、処理中に落ちるとメッセージが失われる false にして、処理完了後に BasicAck
while(true) のビジー ループ コンシューマ側の待機がCPU を食い潰す ManualResetEventSlim 等で待つ
接続の生成 再接続処理が無い AutomaticRecoveryEnabled（既定で有効）
AcceptablePolicyErrors 証明書名の不一致とチェーンのエラーを許容している 本番では許容しない（検証用の設定）
例外処理 Publish のみ try-catch Consumer 側にも必要

特に 1 つ目の autoAck: true は、
Azure Service Busの補足で述べた
ピーク ロックの逆で、**「受け取ったら消える」**方式である。
メッセージの消失が許されない用途では必ず false にする。

channel.BasicConsume(queueName, autoAck: false, consumer);

consumer.Received += (_, ea) =>
{
    try
    {
        // 業務処理
        channel.BasicAck(ea.DeliveryTag, multiple: false);   // 成功したら ACK
    }
    catch
    {
        channel.BasicNack(ea.DeliveryTag, false, requeue: true); // 失敗したら戻す
    }
};

また、AcceptablePolicyErrors で証明書エラーを握り潰すと、
TLS を使っている意味が半減する（中間者攻撃を検知できない）。
検証用の設定であることを明記しておきたい。

EasyNetQ

  • RabbitMQ のための優れた .NET API
  • 初期の開発は、15below がスポンサーとなって行われた。

補足: EasyNetQ は
RabbitMQ.Client を包む高水準ライブラリである。
上のサンプル コードで手書きしていた
Exchange 宣言・Queue 宣言・Binding・シリアライズを
型から自動的に決めてくれる（規約による構成）。

RabbitMQ.Client EasyNetQ
抽象度 低い（AMQP をそのまま） 高い（型でやり取り）
制御 細かく効く 規約に従う
学習 AMQP の理解が要る すぐ書ける
向く場面 細かい制御が要る、既存の Exchange 構成に合わせる .NET 同士の連携を手早く

ただし、Exchange 名や routing key の付け方が
EasyNetQ の規約に従うため、

  • 他言語のアプリと連携する、
  • 既存の Exchange 構成に載せる

といった場合は素の RabbitMQ.Client の方が扱いやすい。

AMQPクライアント

RabbitMQ はサポートしない？

  • AMQPNetLite
  • Microsoft.Azure.Amqp
  • ...

補足（「サポートしない？」への回答）: 疑問形で終わっているので補足する。

結論としては、これらは AMQP 1.0 のクライアントであり、
RabbitMQ は既定では AMQP 1.0 を話さないため、
そのままでは接続できない。

RabbitMQ（既定）  : AMQP 0-9-1（5672 番）
AMQPNetLite       : AMQP 1.0
      ↓ 互換性が無い（名前は同じでも別のプロトコル）
  接続できない

ただし、プラグインで AMQP 1.0 に対応させることは可能である。

rabbitmq-plugins enable rabbitmq_amqp1_0

なお、RabbitMQ 4.0（2024 年）でネイティブに AMQP 1.0 をサポートする
ようになり、状況は改善している。

実務上の指針としては、

接続先 使うライブラリ
RabbitMQ RabbitMQ.Client（AMQP 0-9-1）
Azure Service Bus Azure.Messaging.ServiceBus（AMQP 1.0）
汎用の AMQP 1.0 ブローカー AMQPNetLite

となり、「AMQP だから相互に繋がる」と考えないことが重要である。
Azure Service Busへの移行を検討する際も、
プロトコルが違うためクライアント コードの書き換えが必要になる。

チュートリアル

Windows

証明書の準備

Mosquitto のチュートリアルと同じ。

SSLの設定

  • サーバー証明書を利用

    • advanced.config

      [
  {rabbit,  [
    {ssl_listeners, [5671]},
    {ssl_options, [{cacertfile,"c:/certs/ca.crt"},
                  {certfile,"c:/certs/server.crt"},
                  {keyfile,"c:/certs/server.key"},
                  {verify,verify_none},
                  {fail_if_no_peer_cert,false}]}
  ]}
].

    • サービスの再起動
      管理画面で SSL ポートを確認する。

      Listening ports
amqp/ssl	0.0.0.0	5671
amqp/ssl	::	5671

  • パスワード認証
    管理画面でユーザーを作成する。

  • クライアント証明書を利用

    • プラグインを有効化

      rabbitmq-plugins enable rabbitmq_auth_mechanism_ssl

    • CN と一致したユーザーを作成する。

    • advanced.config

      [
  {rabbit,  [
    {ssl_listeners, [5671]},
    {auth_mechanisms, ['EXTERNAL', 'PLAIN']},
    {ssl_cert_login_from, common_name},
    {ssl_options, [{cacertfile,"c:/certs/ca.crt"},
                  {certfile,"c:/certs/server.crt"},
                  {keyfile,"c:/certs/server.key"},
                  {verify,verify_peer},
                  {fail_if_no_peer_cert,true}]}
  ]}
].

    • サービスの再起動
      管理画面で SSL ポートを確認する。

      Listening ports
amqp/ssl	0.0.0.0	5671
amqp/ssl	::	5671

補足（2 つの設定の差分が「クライアント認証の有無」）: 2 つの
advanced.config を見比べると、違いは 4 箇所である。

項目 サーバー証明書のみ クライアント証明書も
auth_mechanisms （記載なし＝PLAIN 等） ['EXTERNAL', 'PLAIN']
ssl_cert_login_from － common_name（CN をユーザ名とする）
verify verify_none verify_peer（クライアント証明書を検証する）
fail_if_no_peer_cert false true（証明書が無ければ拒否）

つまり、verify_peer ＋ fail_if_no_peer_cert = true の組が
「クライアント証明書を必須にする」という指定である。

verify_none                     … クライアント証明書を見ない
verify_peer + fail_if_...=false … あれば検証、無くても通す（中途半端）
verify_peer + fail_if_...=true  … 【必須】。無ければ接続を拒否 ← これを使う

中間の設定（false）は避けるべきである。
「証明書を出さないクライアントが素通りする」ため、
制御として意味をなさない。

なお、SSL を有効にしても
平文ポート（5672）は開いたままである点に注意する。
閉域でない環境では、listeners.tcp を無効化して
TLS のみ受け付ける構成にする。

プログラムからアクセス

  • 認証局の証明書（ca.crt）は、証明書ストアの
    信頼されたルート証明機関に入れておく必要がある。
  • サーバー証明書を利用

    • 以下を書き換える。

      using System.Net.Security; // 追加

...

// ファクトリ生成
var factory = new ConnectionFactory()
{
  HostName = hostname,
  Ssl = new SslOption // 追加
  {
    Enabled = true,
    ServerName = hostname,
    AcceptablePolicyErrors = SslPolicyErrors.RemoteCertificateNameMismatch
                             | SslPolicyErrors.RemoteCertificateChainErrors,
  }
};

    • デバッグ実行で動作確認
      動作した。

  • パスワード認証を利用

    • 以下を書き換える。

      var factory = new ConnectionFactory()
{
  HostName = hostname,

  UserName = "xxxxx", // 追加
  Password = "xxxxx", // 追加

    • デバッグ実行で動作確認
      動作した。

  • クライアント証明書を利用

    • 以下を書き換える。

      using System.Security.Cryptography.X509Certificates; // 追加

...

X509Certificate2 cli = new X509Certificate2(@"C:\certs\client.pfx", "xxxxx"); // 追加

// ファクトリ生成
var factory = new ConnectionFactory()
{
  HostName = hostname,
  //UserName = "xxxxx", // 削除
  //Password = "xxxxx", // 削除
  AuthMechanisms = new IAuthMechanismFactory[]{ new ExternalMechanismFactory()}, // 追加
  Ssl = new SslOption
  {
    Enabled = true,
    ServerName = hostname,
    AcceptablePolicyErrors = SslPolicyErrors.RemoteCertificateNameMismatch
                             | SslPolicyErrors.RemoteCertificateChainErrors,
    Certs = new X509CertificateCollection(new X509Certificate[] { cli }) // 追加
  }
};

    • デバッグ実行で動作確認
      動作した（ssl_cert_login_from を書く位置に注意！）。

補足（末尾の注意書きについて）: 「ssl_cert_login_from を書く位置に
注意！」という指摘は、Erlang の設定ファイルの構造に起因する。

[
  {rabbit,  [
    {ssl_listeners, [5671]},
    {auth_mechanisms, ['EXTERNAL', 'PLAIN']},
    {ssl_cert_login_from, common_name},   ← 【rabbit 直下】に書く
    {ssl_options, [ ...                    ← ssl_options の【中ではない】
                   ]}
  ]}
].

名前に ssl_ が付くため ssl_options の中に書きたくなるが、
rabbit の直下が正しい。
誤った位置に書いてもエラーにならず、単に無視されるため、
「設定したのに CN でログインできない」という状態になる。
気付きにくい種類の誤りであり、注意書きの価値が高い。

なお、クライアント側の
AuthMechanisms = ... ExternalMechanismFactory() は
**「パスワードではなく証明書で認証する」**という宣言で、
サーバ側の auth_mechanisms に 'EXTERNAL' が含まれている必要がある。
両側の設定が対になっている点を押さえておきたい。

Linux

移行メモ: 原典でも見出しのみで、本文が存在しない。

参考

SIOS Tech. Lab

Microsoft Docs

Qiita

補足（Microsoft Docs の 1 件目が示す使い分け）: 参考の
**「開発環境またはテスト環境の RabbitMQ でイベント バスを実装する」**という
表題が示唆的である。

Microsoft の eShopOnContainers（マイクロサービスの参照実装）では、

【開発・テスト】 RabbitMQ（コンテナで手軽に立てられる）
      ↓ 同じインターフェイスの実装を差し替え
【本番】        Azure Service Bus（マネージド）

という構成を採っている。

環境 選択 理由
開発・テスト RabbitMQ docker run 一発。費用ゼロ、オフラインでも動く
本番 Azure Service Bus 運用不要、SLA、企業向け機能

これを成立させるには、
アプリ側でメッセージ基盤を抽象化しておく必要がある
（IEventBus インターフェイスを定義し、実装を 2 つ用意する）。

クラウド アプリケーション アーキテクチャ ガイドの
「改良を見込んだ設計」（抽象インフラストラクチャをドメイン ロジックと分離）が
そのまま当てはまる実践例である。

Tags: 移行, 通信技術, .NET開発, IoT

このWikiは「Open棟梁Project」,「OSSコンソーシアム 開発基盤部会」によって運営されています。

NetDevInfraWiki

マイクロソフト系技術情報 Wiki
Open 棟梁 Wiki

（未着手）

開発基盤部会 Wiki

移行管理: DONE ／ TODO

Clone this wiki locally