はじめに

RabbitMQによるメッセージのやり取りを行うシステムに関わっているものの体系的な知識がなかったため、RabbitMQ公式のチュートリアルを触ってみました。
https://www.rabbitmq.com/getstarted.html

そこで得た知識と追加で調べた情報などを記載します。

想定読者

RabbitMQをこれから使う人
RabbitMQを使っているけどよくわかってない人
記憶を失った時の自分

RabbitMQとは

RabbitMQはオープンソースのメッセージブローカー。
ブローカー(Broker)は直訳で「仲立ち人」。

複数のメッセージングプロトコルに対応しているがコアとなるのはAMQP。

登場する概念

名前	説明
Producer	メッセージを送るプログラム(＝Publisher)
Exchange	Producerのメッセージはここに送られる。後述のtypeに応じてQueueへ送信
Queue	メッセージを格納する場所
routing key	ExchangeがQueueをbindするために指定される値
Consumer	メッセージを受け取るプログラム(＝Subscriber)

Exchange Type

Exchangeには複数のタイプがある。

fanout

受信したメッセージを全てのQueueにブロードキャストする。

direct

bindする時に指定するrouting key(binding key)がメッセージのrouting keyと一致するQueueに送信する。
言い換えると、routing keyによって送信するQueueを指定する。

topic

複数の条件に対応するため . 区切りで複数の単語を指定する。syslogの「facility.severity」みたいに。
* と # でワイルドカード的な指定ができる。

チュートリアル内容

自分が関わるシステムがExchange type=direct でrouting_keyによるQueueの指定をしていることもあり、それに近い「4 Routing」について書きます。
言語はRubyを選択。bunny gem を読み込んで実行する形でした。

やること

ログレベルをrouting_keyとしてそれに応じて送信するQueueを振り分ける。
図はRabbitMQ公式より

手順

公式のDockerイメージがあるので、まずはそれでブローカーを起動

$ docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.10-management

以下、コード
正直、公式側を見た方がいいのでざっくり。

Publisher( emit_log_direct.rb )

#!/usr/bin/env ruby
require 'bunny'

# コネクション開始
connection = Bunny.new
connection.start

# Channelの作成
channel = connection.create_channel

# Exchangeの指定
exchange = channel.direct('direct_logs')

# 引数をいい感じにする
severity = ARGV.shift || 'info'
message = ARGV.empty? ? 'Hello World!' : ARGV.join(' ')

# Publish!
exchange.publish(message, routing_key: severity)
puts " [x] Sent '#{message}'"

# コネクションクローズ
connection.close

Subscriber( receive_logs_direct.rb )

#!/usr/bin/env ruby
require 'bunny'

abort "Usage: #{$PROGRAM_NAME} [info] [warning] [error]" if ARGV.empty?

# コネクション開始
connection = Bunny.new
connection.start

# Channel作成
channel = connection.create_channel

# Exchange指定
exchange = channel.direct('direct_logs')

# Queueの宣言。冪等で、ない場合は作成されるらしい
# exclusive: true でコネクション終了時にQueueを削除する
queue = channel.queue('', exclusive: true)

# 実行(Subscriber起動)時に引数で指定するログレベルごとにQueueをbind
ARGV.each do |severity|
  queue.bind(exchange, routing_key: severity)
end

puts ' [*] Waiting for logs. To exit press CTRL+C'

# メッセージが送信されるとsubscribeメソッドが実行される
begin
  queue.subscribe(block: true) do |delivery_info, _properties, body|
    puts " [x] #{delivery_info.routing_key}:#{body}"
  end
rescue Interrupt => _
  channel.close
  connection.close

  exit(0)
end

Subscriberプログラムを起動して

$ ruby receive_logs_direct.rb info error warning
 [*] Waiting for logs. To exit press CTRL+C

Publisherプログラム実行

$ ruby emit_log_direct.rb error "Aka---n!"
 [x] Sent 'Aka---n!'

やったぜ。

$ ruby receive_logs_direct.rb info error warning
 [*] Waiting for logs. To exit press CTRL+C
 [x] error:Aka---n!

上記の図のように、引数として指定するログレベルを変えて(routing_keyを変えて)複数のSubscriberを立ち上げても意図した通りに動く。

$ ruby emit_log_direct.rb warning "Akankamoshiren"
 [x] Sent 'Akankamoshiren'

それぞれ別ターミナルで起動したSubscriber↓

$ ruby receive_logs_direct.rb error
 [*] Waiting for logs. To exit press CTRL+C

$ ruby receive_logs_direct.rb info error warning
 [*] Waiting for logs. To exit press CTRL+C
 [x] warning:Akankamoshiren

おまけ

Virtual Hosts

RabbitMQではVirtual Hosts(vhost)としてリソースを論理的なグループに分離することができる。
https://www.rabbitmq.com/vhosts.html

Namespace的なものだと解釈した。