mkdir rabbitmq-test && cd rabbitmq-test

mise use python@3.12

cat << 'EOS' >> .mise.toml

[env]
_.python.venv = { path = ".venv", create = true }
EOS

docker run -it --rm \
    --name rabbitmq \
    -p 5672:5672 \
    -p 15672:15672 \
    rabbitmq:4.0-management

pip install pika

pip freeze | grep -i pika

pika==1.3.2

import pika

# RabbitMQサーバへ接続
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
# メッセージ送受信用のチャネルを開く
channel = connection.channel()

# キュー名の宣言
channel.queue_declare(queue='hello')

# メッセージをキューに送信（publish）
channel.basic_publish(
    # ルーティング時に使用する（今回は未使用）
    exchange='',
    # メッセージ送信先キュー名
    routing_key='hello',
    # 送信するメッセージ
    body='こんにちは！'
)
print(" [x] 'こんにちは！' を送信しました。")

# RabbitMQサーバとの接続をクローズ
connection.close()

import pika, sys, os

def main():
    # RabbitMQサーバへ接続
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('localhost')
    )
    # メッセージ送受信用のチャネルを開く
    channel = connection.channel()
    # キュー名の宣言
    channel.queue_declare(queue='hello')

    # メッセージ受診時に呼び出されるコールバック関数
    def callback(ch, method, properties, body):
        print(f" [x] '{body}' を受信しました。")

    # メッセージを受信(消費)
    channel.basic_consume(
        # 対象のキュー名
        queue='hello',
        # キュー受信時に呼び出すコールバック関数
        on_message_callback=callback,
        # メッセージ受信後、RabbitMQに自動的に確認（acknowledge）を送信->そのキューを削除
        auto_ack=True
    )
    print(' [*] メッセージを待っています。終了する場合は CTRL+C で終了します。')

    # メッセージ受信を開始（無限ループで待機）
    channel.start_consuming()

if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

python receive.py

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。

python send.py

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
 [x] 'b'\xe3\x81\x93\xe3\x82\x93\xe3\x81\xab\xe3\x81\xa1\xe3\x81\xaf\xef\xbc\x81'' を受信しました。

(snip)
    def callback(ch, method, properties, body):
        # メッセージをUTF−8でデコード
        decoded_body = body.decode('utf-8')
        print(f" [x] '{decoded_body}' を受信しました。")
(snip)

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
 [x] 'こんにちは！' を受信しました。
 [x] 'こんにちは！' を受信しました。
 [x] 'こんにちは！' を受信しました。

python send.py
python send.py
python send.py
python send.py

docker exec -ti rabbitmq rabbitmqctl list_queues

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages
hello	4

python receive.py

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
 [x] 'こんにちは！' を受信しました。
 [x] 'こんにちは！' を受信しました。
 [x] 'こんにちは！' を受信しました。
 [x] 'こんにちは！' を受信しました。

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()
channel.queue_declare(queue='hello')

# コマンドライン引数でメッセージを取得
message = ' '.join(sys.argv[1:]) or "こんにちは！"

channel.basic_publish(
    exchange='',
    routing_key='hello',
    body=message
)
print(f" [x] '{messasge}' を送信しました。")

connection.close()

import pika, sys, os
import time
from datetime import datetime

def main():
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('localhost')
    )
    channel = connection.channel()
    channel.queue_declare(queue='hello')

    def callback(ch, method, properties, body):
        decoded_body = body.decode('utf-8')
        print("{} [x] '{}' を受信しました。".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'), decoded_body))
        # 時間がかかる処理を擬似的に再現するためのスリープ
        time.sleep(len(decoded_body))
        print("{} [x] '{}' を完了しました。".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'), decoded_body))

    channel.basic_consume(
        queue='hello',
        on_message_callback=callback,
        auto_ack=True
    )
    print(' [*] メッセージを待っています。終了する場合は CTRL+C で終了します。')

    channel.start_consuming()

if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

python worker.py

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。

python worker.py

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。

python new_task.py "テスト1"
python new_task.py "テスト2"
python new_task.py "テスト3"
python new_task.py "テスト4"

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 16:24:53 [x] 'テスト1' を受信しました。
2025-01-26 16:24:57 [x] 'テスト1' を完了しました。
2025-01-26 16:24:57 [x] 'テスト3' を受信しました。
2025-01-26 16:25:01 [x] 'テスト3' を完了しました。

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 16:24:54 [x] 'テスト2' を受信しました。
2025-01-26 16:24:58 [x] 'テスト2' を完了しました。
2025-01-26 16:24:58 [x] 'テスト4' を受信しました。
2025-01-26 16:25:02 [x] 'テスト4' を完了しました。

(snip)
    channel.basic_consume(
        queue='hello',
        on_message_callback=callback,
        auto_ack=True  # メッセージ受信したら、確認応答を自動で返す、つまりタスク成功可否に関係ない
    )
(snip)

(snip)
    def callback(ch, method, properties, body):
        decoded_body = body.decode('utf-8')
        print("{} [x] '{}' を受信しました。".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'), decoded_body))
        time.sleep(len(decoded_body))
        print("{} [x] '{}' を完了しました。".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'), decoded_body))
        # 処理成功後に手動でackを返す
        ch.basic_ack(delivery_tag = method.delivery_tag)

    channel.basic_consume(
        queue='hello',
        on_message_callback=callback,
        auto_ack=False  # auto_ackを無効化。デフォルトは無効なので指定しなくてもよい。
    )
(snip)

python worker.py

python worker.py

python new_task.py "テストテストテスト1"

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 16:51:29 [x] 'テストテストテスト1' を受信しました。
2025-01-26 16:51:38 [x] 'テストテストテスト1' を完了しました。

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。

python new_task.py "テストテストテスト2"

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 16:52:38 [x] 'テストテストテスト2' を受信しました。

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 16:52:38 [x] 'テストテストテスト2' を受信しました。
^CInterrupted

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 16:51:29 [x] 'テストテストテスト1' を受信しました。
2025-01-26 16:51:38 [x] 'テストテストテスト1' を完了しました。
2025-01-26 16:52:41 [x] 'テストテストテスト2' を受信しました。
2025-01-26 16:52:51 [x] 'テストテストテスト2' を完了しました。

docker exec -ti rabbitmq rabbitmqctl list_queues  name messages_ready messages_unacknowledged

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages_ready	messages_unacknowledged
hello	0	0

2025-01-26 17:01:29 [x] 'テストテストテスト3' を受信しました。

docker exec -ti rabbitmq rabbitmqctl list_queues  name messages_ready messages_unacknowledged

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages_ready	messages_unacknowledged
hello	0	1

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages_ready	messages_unacknowledged
hello	1	0

(snip)
channel.queue_declare(
    queue='hello',
    durable=True    # キューを「耐久性あり」に指定
)
(snip)

pika.exceptions.ChannelClosedByBroker: (406, "PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'hello' in vhost '/': received 'true' but current is 'false'")

^C2025-01-26 08:27:47.461149+00:00 [notice] <0.64.0> SIGTERM received - shutting down
(snip)
2025-01-26 08:27:47.495199+00:00 [info] <0.555.0> Message store for directory '/var/lib/rabbitmq/mnesia/rabbit@3645d298b22d/msg_stores/vhosts/628WB79CIFDYO9LJI6DKMI09L/msg_store_transient' is stopped

docker run -it -d \
    --name rabbitmq \
    -h "rabbitmq" \
    -p 5672:5672 \
    -p 15672:15672 \
    -v ./rabbitmq-data:/var/lib/rabbitmq/mnesia \
    rabbitmq:4.0-management

python new_task.py "テスト"

docker exec -ti rabbitmq rabbitmqctl list_queues

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages
hello	1

docker restart rabbitmq

docker exec -ti rabbitmq rabbitmqctl list_queues

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages
hello	0

(snip)
channel.queue_declare(queue='hello', durable=True)
channel.queue_declare(queue='non_durable_hello')
(snip)

(snip)
channel.basic_publish(
    exchange='',
    routing_key='hello',
    body=message,
    # メッセージのプロパティで「耐久性あり」を指定
    properties=pika.BasicProperties(
       delivery_mode = pika.DeliveryMode.Persistent
    )
)
(snip)

python new_task.py "テスト"

docker exec -ti rabbitmq rabbitmqctl list_queues

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages
hello	1
non_durable_hello	0

docker restart rabbitmq

docker exec -ti rabbitmq rabbitmqctl list_queues

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages
hello	1

python new_task.py "テストテストテストテストテスト"
python new_task.py "テスト"
python new_task.py "テストテストテストテストテスト"
python new_task.py "テスト"
python new_task.py "テストテストテストテストテスト"
python new_task.py "テスト"

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 18:31:04 [x] 'テストテストテストテストテスト' を受信しました。
2025-01-26 18:31:19 [x] 'テストテストテストテストテスト' を完了しました。
2025-01-26 18:31:19 [x] 'テストテストテストテストテスト' を受信しました。
2025-01-26 18:31:34 [x] 'テストテストテストテストテスト' を完了しました。
2025-01-26 18:31:34 [x] 'テストテストテストテストテスト' を受信しました。
2025-01-26 18:31:49 [x] 'テストテストテストテストテスト' を完了しました。

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 18:31:05 [x] 'テスト' を受信しました。
2025-01-26 18:31:08 [x] 'テスト' を完了しました。
2025-01-26 18:31:08 [x] 'テスト' を受信しました。
2025-01-26 18:31:11 [x] 'テスト' を完了しました。
2025-01-26 18:31:11 [x] 'テスト' を受信しました。
2025-01-26 18:31:14 [x] 'テスト' を完了しました。

    # ワーカーが同時受信可能なメッセージ数を1に制限
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(
        queue='hello',
        on_message_callback=callback,
        auto_ack=False
    )
    print(' [*] メッセージを待っています。終了する場合は CTRL+C で終了します。')

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 18:42:19 [x] 'テストテストテストテストテスト' を受信しました。
2025-01-26 18:42:34 [x] 'テストテストテストテストテスト' を完了しました。
2025-01-26 18:42:34 [x] 'テスト' を受信しました。
2025-01-26 18:42:37 [x] 'テスト' を完了しました。
2025-01-26 18:42:37 [x] 'テストテストテストテストテスト' を受信しました。
2025-01-26 18:42:52 [x] 'テストテストテストテストテスト' を完了しました。

 [*] メッセージを待っています。終了する場合は CTRL+C で終了します。
2025-01-26 18:42:21 [x] 'テスト' を受信しました。
2025-01-26 18:42:24 [x] 'テスト' を完了しました。
2025-01-26 18:42:24 [x] 'テストテストテストテストテスト' を受信しました。
2025-01-26 18:42:39 [x] 'テストテストテストテストテスト' を完了しました。
2025-01-26 18:42:39 [x] 'テスト' を受信しました。
2025-01-26 18:42:42 [x] 'テスト' を完了しました。

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# エクスチェンジの宣言
channel.exchange_declare(
    exchange='logs',
    exchange_type='fanout'
)

message = ' '.join(sys.argv[1:]) or "info: こんにちは！"

channel.basic_publish(
    exchange='logs',  # エクスチェンジを指定
    routing_key='',
    body=message,
)
print(f" [x] ログ '{message}' を送信しました。")

connection.close()

channel.basic_publish(
    exchange='',
    routing_key='hello',
    body='こんにちは！'
)

channel.exchange_declare(
    exchange='logs',        # エクスチェンジの名前
    exchange_type='fanout'  # エクスチェンジの種類
)

(snip)
channel.basic_publish(
    exchange='logs',  # エクスチェンジを指定
    routing_key='',
    body=message,
)
(snip)

docker exec -ti rabbitmq rabbitmqctl list_exchanges

Listing exchanges for vhost / ...
name	type
amq.direct	direct
amq.topic	topic
amq.fanout	fanout
amq.match	headers
	direct
amq.rabbitmq.trace	topic
amq.headers	headers

python emit_log.py "ログ1"

Listing exchanges for vhost / ...
name	type
amq.direct	direct
amq.topic	topic
amq.fanout	fanout
amq.match	headers
	direct
amq.rabbitmq.trace	topic
amq.headers	headers
logs	fanout

docker exec -ti rabbitmq rabbitmqctl list_queues

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name	messages
hello	0
non_durable_hello	0

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# エクスチェンジの宣言
channel.exchange_declare(exchange='logs', exchange_type='fanout')

# 一時的な名前付きキューを作成#
result = channel.queue_declare(
    queue='',      # 空文字列の場合、ランダムな名前付きキューが自動生成される
    exclusive=True # 現在の接続でのみ使用され、接続が閉じると削除される一時キューになる
)
# キュー名を取得する
queue_name = result.method.queue
print("[DEBUG] Queue: {queue_name}")

# キュー名とエクスチェンジを紐づけ
channel.queue_bind(exchange='logs', queue=queue_name)

print(' [*] ログを待っています。終了する場合は CTRL+C で終了します。')

def callback(ch, method, properties, body):
    decoded_body = body.decode('utf-8')
    print(f" [x] {decoded_body}")

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

channel.start_consuming()

(snip)
# 一時的な名前付きキューを作成#
result = channel.queue_declare(
    queue='',      # 空文字列の場合、ランダムな名前付きキューが自動生成される
    exclusive=True # 現在の接続でのみ使用され、接続が閉じると削除される一時キューになる
)
# ランダムに生成されたキュー名を取得する
queue_name = result.method.queue
(snip)

# キュー名とエクスチェンジを紐づけ
channel.queue_bind(exchange='logs', queue=queue_name)

docker exec -ti rabbitmq rabbitmqctl list_bindings

Listing bindings for vhost /...
source_name	source_kind	destination_name	destination_kind	routing_key	arguments
	exchange	hello	queue	hello	[]
	exchange	non_durable_hello	queue	non_durable_hello	[]

python receive_log.py

[DEBUG] Queue: amq.gen-ByfpokCJnrbbYKXKejZVQQ
 [*] ログを待っています。終了する場合は CTRL+C で終了します。

python receive_log.py

[DEBUG] Queue: amq.gen-K1Us9NRxxwVlqdOMXafmXQ
 [*] ログを待っています。終了する場合は CTRL+C で終了します。

python emit_log.py "ログ1"
python emit_log.py "ログ2"
python emit_log.py "ログ3"

[DEBUG] Queue: amq.gen-ByfpokCJnrbbYKXKejZVQQ
 [*] ログを待っています。終了する場合は CTRL+C で終了します。
 [x] ログ1
 [x] ログ2
 [x] ログ3

[DEBUG] Queue: amq.gen-K1Us9NRxxwVlqdOMXafmXQ
 [*] ログを待っています。終了する場合は CTRL+C で終了します。
 [x] ログ1
 [x] ログ2
 [x] ログ3

docker exec -ti rabbitmq rabbitmqctl list_bindings

Listing bindings for vhost /...
source_name	source_kind	destination_name	destination_kind	routing_key	arguments
	exchange	hello	queue	hello	[]
	exchange	amq.gen-K1Us9NRxxwVlqdOMXafmXQ	queue	amq.gen-K1Us9NRxxwVlqdOMXafmXQ	[]
	exchange	amq.gen-ByfpokCJnrbbYKXKejZVQQ	queue	amq.gen-ByfpokCJnrbbYKXKejZVQQ	[]
	exchange	non_durable_hello	queue	non_durable_hello	[]
logs	exchange	amq.gen-ByfpokCJnrbbYKXKejZVQQ	queue	amq.gen-ByfpokCJnrbbYKXKejZVQQ	[]
logs	exchange	amq.gen-K1Us9NRxxwVlqdOMXafmXQ	queue	amq.gen-K1Us9NRxxwVlqdOMXafmXQ	[]

channel.queue_bind(
    exchange=exchange_name,  # メッセージを送信するエクスチェンジ
    queue=queue_name         # メッセージを受け取るキュー
)

channel.queue_bind(
    exchange=exchange_name,  # メッセージを送信するエクスチェンジ
    queue=queue_name,        # メッセージを受け取るキュー
    routing_key='black'      # バインディングキーの設定
)

channel.basic_publish(
    exchange='direct_logs',   # 送信先のエクスチェンジ
    routing_key='orange',     # メッセージのルーティングキー
    body='...'                # メッセージの内容
)

# エクスチェンジの宣言
channel.exchange_declare(
    exchange='direct_logs',
    exchange_type='direct'  # エクスチェンジの種類を`direct`に指定
)

# キューの宣言
channel.queue_declare(queue='orange_queue')

# エクスチェンジとキューをバインディング
channel.queue_bind(
    exchange='direct_logs',  # メッセージを送信するエクスチェンジ
    queue="orange_queue",    # メッセージを受け取るキュー
    routing_key='orange'     # バインディングキーの設定
)

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

channel.queue_declare(queue='orange_queue')
channel.queue_declare(queue='other_queue')

channel.queue_bind(
    exchange='direct_logs',
    queue="orange_queue"
    routing_key='orange'
)
channel.queue_bind(
    exchange='direct_logs',
    queue="other_queue",
    routing_key='green'
)
channel.queue_bind(
    exchange='direct_logs',
    queue="other_queue",
    routing_key='black'
)

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

channel.queue_declare(queue='q1')
channel.queue_declare(queue='q2')

channel.queue_bind(
    exchange='direct_logs',
    queue="q1"
    routing_key='black'
)
channel.queue_bind(
    exchange='direct_logs',
    queue="q2",
    routing_key='black'
)

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

result = channel.queue_declare(
    queue='', exclusive=True
)
queue_name = result.method.queue

# severities（ログレベル）は起動時のコマンドライン引数から取得する
for severity in severities:
    # ログレベルごとに、ログレベルがルーティングキーに指定されたバインディングを設定
    channel.queue_bind(
        exchange='direct_logs',
        queue=queue_name,
        routing_key=severity
)

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

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

severity = sys.argv[1] if len(sys.argv) > 1 else "info"
message = ' '.join(sys.argv[2:]) or "テストログ"

channel.basic_publish(
    exchange='direct_logs',
    routing_key=severity,
    body=message,
)
print(f" [x] ログ '{severity}:{message}' を送信しました。")

connection.close()

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

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

result = channel.queue_declare(
    queue='',
    exclusive=True
)

queue_name = result.method.queue

severities = sys.argv[1:]
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
    sys.exit(1)

for severity in severities:
    channel.queue_bind(
        exchange='direct_logs',
        queue=queue_name,
        routing_key=severity
    )

print(' [*] ログを待っています。終了する場合は CTRL+C で終了します。')

def callback(ch, method, properties, body):
    decoded_body = body.decode('utf-8')
    print(f" [x] {method.routing_key}:{decoded_body}")

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

channel.start_consuming()

PYTHONUNBUFFERED=1 python receive_log_direct.py warning error > logs_from_rabbit.log &
tail -f logs_from_rabbit.log

python receive_log_direct.py info warning error

python emit_log_direct.py info "infoログ"

 [*] ログを待っています。終了する場合は CTRL+C で終了します。

 [*] ログを待っています。終了する場合は CTRL+C で終了します。
 [x] info:ログ1

python emit_log_direct.py error "errorログ"
python emit_log_direct.py warning "warningログ"

 [*] ログを待っています。終了する場合は CTRL+C で終了します。
 [x] error:errorログ
 [x] warning:warningログ

 [*] ログを待っています。終了する場合は CTRL+C で終了します。
 [x] info:ログ1
 [x] error:errorログ
 [x] warning:warningログ

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

channel.exchange_declare(
    exchange='topic_logs',
    exchange_type='topic' # topicタイプを指定
)

# ルーティングキーは <facility>.<severity> を想定
routing_key = sys.argv[1] if len(sys.argv) > 2 else "anonymous.info"
message = ' '.join(sys.argv[2:]) or "テストログ"

channel.basic_publish(
    exchange='topic_logs',
    routing_key=routing_key,
    body=message,
)
print(f" [x] ログ '{routing_key}:{message}' を送信しました。")

connection.close()

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(
    exchange='topic_logs',
    exchange_type='topic'
)

result = channel.queue_declare(
    queue='',
    exclusive=True
)

queue_name = result.method.queue

binding_keys = sys.argv[1:]
if not binding_keys:
    sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
    sys.exit(1)

for binding_key in binding_keys:
    channel.queue_bind(
        exchange='topic_logs',
        queue=queue_name,
        routing_key=binding_key
    )

print(' [*] ログを待っています。終了する場合は CTRL+C で終了します。')

def callback(ch, method, properties, body):
    decoded_body = body.decode('utf-8')
    print(f" [x] {method.routing_key}:{decoded_body}")

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

channel.start_consuming()

python receive_log_topic.py "#"

python receive_log_topic.py "kern.*"

python receive_log_topic.py "*.critical"

python receive_log_topic.py "kern.*" "*.critical"

python emit_log_topic.py "kern.critical" "クリティカルなカーネルエラー"

 [x] kern.critical:クリティカルなカーネルエラー

python emit_log_topic.py "kern.info" "普通のカーネルログ"

 [x] kern.info:普通のカーネルログ

python emit_log_topic.py "auth.info" "普通の認証ログ"

 [x] auth.info:普通の認証ログ

fibonacci_rpc = FibonacciRpcClient()
result = fibonacci_rpc.call(4)
print(f"fib(4) is {result}")

(snip)

# コールバックキューを作成
result = channel.queue_declare(queue='', exclusive=True)
callback_queue = result.method.queue

# リクエストを送信
channel.basic_publish(
    exchange='',
    routing_key='rpc_queue',  # クライアントからの送信先キュー
    properties=pika.BasicProperties(
        reply_to = callback_queue,    # サーバからの結果の返信先キュー
    ),
    body=request
)
(snip)

# コールバックキューを作成
result = channel.queue_declare(queue='', exclusive=True)
callback_queue = result.method.queue

import pika
import sys

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# rpc_queueを宣言
result = channel.queue_declare(queue='rpc_queue')

# フィボナッチ数を返す関数
def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)

# リクエスト時のコールバック関数
def on_request(ch, method, props, body):
    n = int(body)
    print(f" [.] fib({n})")
    
    # フィボナッチ関数を求める
    response = fib(n)

    # コールバックキューにメッセージを発行 
    ch.basic_publish(
        exchange="",
        routing_key=props.reply_to,               # コールバックキューを指定
        properties=pika.BasicProperties(
            correlation_id = props.correlation_id # correlation_idを指定
        ),
        body=str(response)
    )
    # rpc_queueにackを返す
    ch.basic_ack(delevery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
# rpc_queueのキューを受信して処理
channel.basic_consume(
    queue="rpc_queue",
    on_message_callback=on_request,
)

print(' [*] RPCリクエストを待っています。')
channel.start_consuming()

import pika
import uuid

# フィボナッチ数を求めるRPCクライアントクラス
class FibonacciRpcClient(object):

    def __init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host='localhost'))

        self.channel = self.connection.channel()

        # コールバックキューを宣言
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue

        # コールバックキューをサブスクライブ
        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True
        )

        self.response = None
        self.corr_id = None

    # レスポンス時のコールバック関数
    def on_response(self, ch, method, props, body):
        # correlation_idが一致したら結果を取得
        if self.corr_id == props.correlation_id:
            self.response = body

    # RPCサービスを呼び出す
    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())  # co

        # ｒｐｃ_queueにリクエストを送信
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',
            properties=pika.BasicProperties(
                reply_to=self.callback_queue, # コールバックキューを指定
                correlation_id=self.corr_id,  # correlation_idを指定
            ),
            body=str(n))
        # コールバックキューからのレスポンスが返ってくるまで待つ
        while self.response is None:
            self.connection.process_data_events(time_limit=None)
        return int(self.response)

# クライアント初期化
fibonacci_rpc = FibonacciRpcClient()

# フィボナッチ数を求めるリクエストを送信
print(" [x] リクエストしています: fib(30)")
response = fibonacci_rpc.call(30)
print(f" [.] 結果: {response}")

python rpc_server.py

[*] RPCリクエストを待っています。

python rpc_client.py

 [x] リクエストしています: fib(30)
 [.] 結果: 832040

 [*] RPCリクエストを待っています。
 [.] fib(30)

docker run -it -d \
    --name rabbitmq \
    -h "rabbitmq" \
    -p 5672:5672 \
    -p 15672:15672 \
    -p 5552:5552 \
    -e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS='-rabbitmq_stream advertised_host localhost' \
    -v ./rabbitmq-data:/var/lib/rabbitmq/mnesia \
    rabbitmq:4.0-management

docker exec rabbitmq rabbitmq-plugins enable rabbitmq_stream rabbitmq_stream_management 

Enabling plugins on node rabbit@rabbitmq:
rabbitmq_stream
rabbitmq_stream_management
The following plugins have been configured:
  rabbitmq_management
  rabbitmq_management_agent
  rabbitmq_prometheus
  rabbitmq_stream
  rabbitmq_stream_management
  rabbitmq_web_dispatch
Applying plugin configuration to rabbit@rabbitmq...
The following plugins have been enabled:
  rabbitmq_stream
  rabbitmq_stream_management

started 2 plugins.

pip install rstream

pip freeze | grep -i rstream

rstream==0.20.8

import asyncio

from rstream import Producer

STREAM_NAME = "hello-python-stream"
# 5GB
STREAM_RETENTION = 5000000000


async def send():
    # RabbitMQサーバに接続
    async with Producer(
        host="localhost",
        username="guest",
        password="guest",
    ) as producer:
        # 新しいストリームを作成
        await producer.create_stream(
            # ストリーム名
            STREAM_NAME,
            # ストリームが既に存在する場合は無視
            exists_ok=True,
            # ストリームの最大サイズを5GBに設定
            arguments={"max-length-bytes": STREAM_RETENTION}
        )

        # メッセージを送信
        await producer.send(
            # ストリーム名
            stream=STREAM_NAME,
            # メッセージ
            # 訳注: メッセージはバイト列で送信する必要がある。
            # 元のコードは
            # `message=b'Hello, World!'`
            # となっていたが、日本語の場合は
            # `encode("utf-8")`
            # でバイト列に変換する。
            message="こんにちは！".encode("utf-8")
        )

        print(" [x] メッセージを送信しました。")

        input(" [x] Enterを押してプロデューサを終了します。")

with asyncio.Runner() as runner:
    runner.run(send())

async def send():
    async with Producer(
        host="localhost",
        username="guest",
        password="guest",
    ) as producer:

STREAM_NAME = "hello-python-stream"
# 5GB
STREAM_RETENTION = 5000000000

(snip)
        await producer.create_stream(
            # ストリーム名
            STREAM_NAME,
            exists_ok=True,
            arguments={"max-length-bytes": STREAM_RETENTION}
        )

        await producer.send(
            stream=STREAM_NAME,
            message="こんにちは！".encode("utf-8")
        )

import asyncio

from rstream import (
    AMQPMessage,
    Consumer,
    ConsumerOffsetSpecification,
    MessageContext,
    OffsetType,
)

STREAM_NAME = "hello-python-stream"
# 5GB
STREAM_RETENTION = 5000000000


async def receive():
    # RabbitMQサーバに接続
    async with Consumer(
        host="localhost",
        username="guest",
        password="guest"
    ) as consumer:
        # ストリームを作成
        await consumer.create_stream(
            # ストリーム名
            STREAM_NAME,
            # ストリームが既に存在する場合は無視
            exists_ok=True,
            # ストリームの最大サイズを5GBに設定
            arguments={"max-length-bytes": STREAM_RETENTION}
        )

        # メッセージを受信したときのコールバック関数
        #   msg:
        #     受信したメッセージを表すAMQPMessageオブジェクト
        #   message_context:
        #     メッセージのコンテキストを表すMessageContextオブジェクト
        #     どのストリームからのメッセージかを識別するために使用
        async def on_message(msg: AMQPMessage, message_context: MessageContext):
            # ストリームを取得
            stream = message_context.consumer.get_stream(message_context.subscriber_name)
            # 日本語の場合はデコードする必要がある
            print("ストリーム {} から メッセージを受信: {}".format(stream, msg.decode("utf-8")))

        print("Enterを押してコンシューマを終了します。")
        # コンシュマーを開始
        await consumer.start()
        # ストリームをサブスクライプ
        await consumer.subscribe(
            # ストリーム名
            stream=STREAM_NAME,
            # メッセージを受信したときのコールバック関数
            callback=on_message,
            # ストリーム内のどこからメッセージを受信するかを指定
            # 以下の場合は、最初のメッセージから受信を開始する
            offset_specification=ConsumerOffsetSpecification(OffsetType.FIRST, None),
        )

        try:
            # コンシュマーを無限ループで実行
            await consumer.run()
        except (KeyboardInterrupt, asyncio.CancelledError):
            print("コンシューマを終了します。。。")
            return


with asyncio.Runner() as runner:
    runner.run(receive())

(snip)

STREAM_NAME = "hello-python-stream"
# 5GB
STREAM_RETENTION = 5000000000

async def receive():
    async with Consumer(
        host="localhost",
        username="guest",
        password="guest"
    ) as consumer:
        await consumer.create_stream(
            STREAM_NAME,
            exists_ok=True,
            arguments={"max-length-bytes": STREAM_RETENTION}
        )
(snip)

        async def on_message(msg: AMQPMessage, message_context: MessageContext):
            stream = message_context.consumer.get_stream(message_context.subscriber_name)
            print("ストリーム {} から メッセージを受信: {}".format(stream, msg.decode("utf-8")))

        print("Enterを押してコンシューマを終了します。")
        await consumer.start()
        await consumer.subscribe(
            stream=STREAM_NAME,
            callback=on_message,
            offset_specification=ConsumerOffsetSpecification(OffsetType.FIRST, None),
        )

        try:
            await consumer.run()
        except (KeyboardInterrupt, asyncio.CancelledError):
            print("コンシューマを終了します。。。")
            return

python receive.py

Enterを押してコンシューマを終了します。

python send.py

 [x] メッセージを送信しました。
 [x] Enterを押してプロデューサを終了します。

Enterを押してコンシューマを終了します。
ストリーム hello-python-stream から メッセージを受信: こんにちは！

Enterを押してコンシューマを終了します。
ストリーム hello-python-stream から メッセージを受信: こんにちは！
ストリーム hello-python-stream から メッセージを受信: こんにちは！
ストリーム hello-python-stream から メッセージを受信: こんにちは

python receive.py

Enterを押してコンシューマを終了します。
ストリーム hello-python-stream から メッセージを受信: こんにちは！
ストリーム hello-python-stream から メッセージを受信: こんにちは！
ストリーム hello-python-stream から メッセージを受信: こんにちは！