RabbitMQのRoutingの実験

メッセージをブロードキャストでキューに配信していたものをルーティングキーを使用して、配信をコントロールする。

具体的には、ログレベルの高いメッセージはファイルに出力して、他のメッセージはファイルに出力しないことができる。

www.rabbitmq.com

やり方

routing_keyを分けることで別々のキューにメッセージを配信できる。まずは、exchangeの宣言時に、exchange_typeをdirectに設定する。

channel.exchange_declare(exchange='direct_logs',
                         exchange_type='direct')

Queueとexchangeをバインドする際に、routing_keyを指定する。

channel.queue_bind(exchange=exchange_name,
                   queue=queue_name,
                   routing_key='black')

メッセージの配信時にrouting_keyを指定し、メッセージを配信する。

channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)

ログレベルでコンソールかファイル出力かを分ける

メッセージを受け取る側でdebug/infoレベルのメッセージの場合はログファイルに出力し、warn/errorレベルのメッセージの場合はコンソールでメッセージを出力する。

受信側

ログレベル単位でキューを用意する。キュー単位にコールバック関数も用意。

for severity in severities:
    result = channel.queue_declare(queue="", exclusive=True)
    queue_name = result.method.queue
    channel.queue_bind(exchange="direct_logs", queue=queue_name, routing_key=severity)

    if severity == "debug":

        channel.basic_consume(
            queue=queue_name, on_message_callback=callback_for_debug, auto_ack=True
        )

    if severity == "info":
        channel.basic_consume(
            queue=queue_name, on_message_callback=callback_for_info, auto_ack=True
        )

    if severity == "warn":
        channel.basic_consume(
            queue=queue_name, on_message_callback=callback_for_warn, auto_ack=True
        )

    if severity == "error":
        channel.basic_consume(
            queue=queue_name, on_message_callback=callback_for_error, auto_ack=True
        )

送信側

送信側は、対象キューにメッセージを配信するために、routing_keyを設定する。

messages = [
    {"level": "debug", "message": "Debug Message in console"},
    {"level": "info", "message": "Info Message in console"},
    {"level": "warn", "message": "Warn Message in console"},
    {"level": "error", "message": "Error Message in console"},
]

for message in messages:
    channel.basic_publish(
        exchange="direct_logs", routing_key=message["level"], body=message["message"]
    )
    print("Send " + message["message"])

動作結果

コンソール出力とファイル出力に分けることが出来た。

# コンソール
 [*] Waiting for logs. To exit press CTRL+C
-- debug callback ----
-- info callback ----
-- warn callback ----
2024-04-23 07:19:14,171 -  [x] warn:b'Warn Message in console'
-- error callback ----
2024-04-23 07:19:15,176 -  [x] error:b'Error Message in console'

# ファイル
DEBUG  2024-04-23 07:19:12,159  [__main__]  [x] debug:b'Debug Message in console'
INFO  2024-04-23 07:19:13,165  [__main__]  [x] info:b'Info Message in console'