MS_RabbitMQ - NetDevInfraWGinOSSConsortium/NetDevInfraWiki GitHub Wiki

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コン゜ヌシアム 開発基盀郚䌚」によっお運営されおいたす。

⚠ **GitHub.com Fallback** ⚠