RabbitMQ（消息代理）

# 安装 pika（RabbitMQ 的 Python 客户端）
pip install pika==1.3.2

# 用 Docker 一键启动 RabbitMQ（含 Web 管理界面）
docker run -d --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  rabbitmq:4-management

import json
import pika

params = pika.ConnectionParameters(
    host="localhost",
    port=5672,
    connection_attempts=3,
    retry_delay=1,
)

try:
    connection = pika.BlockingConnection(params)
except pika.exceptions.AMQPConnectionError:
    print("[提示] RabbitMQ 未启动或 5672 端口不可用，请先执行上面的 Docker 启动命令。")
else:
    channel = connection.channel()

    # 2. 声明队列（不存在则自动创建，durable=True 开启持久化）
    channel.queue_declare(queue="hello_queue", durable=True)

    # 3. 发送 3 条消息
    for i in range(3):
        msg = {"id": i, "content": f"第 {i} 条消息"}
        channel.basic_publish(
            exchange="",                # 空字符串 = 使用默认交换器
            routing_key="hello_queue",  # 默认交换器下，路由键就是队列名
            body=json.dumps(msg, ensure_ascii=False).encode("utf-8"),
            properties=pika.BasicProperties(delivery_mode=2),  # 消息持久化
        )
        print(f"[已发送] {msg}")

    connection.close()

import json
import pika

params = pika.ConnectionParameters(
    host="localhost",
    port=5672,
    connection_attempts=3,
    retry_delay=1,
)

try:
    connection = pika.BlockingConnection(params)
except pika.exceptions.AMQPConnectionError:
    print("[提示] RabbitMQ 未启动或 5672 端口不可用，请先执行上面的 Docker 启动命令。")
else:
    channel = connection.channel()

    # 2. 声明同一个队列（消费者也要声明，防止队列不存在）
    channel.queue_declare(queue="hello_queue", durable=True)

    print("[消费者] 开始拉取消息，连续 2 轮没有新消息就退出")
    idle_rounds = 0

    for method_frame, properties, body in channel.consume(
        "hello_queue", inactivity_timeout=1
    ):
        if method_frame is None:
            idle_rounds += 1
            if idle_rounds >= 2:
                print("[消费者] 暂无新消息，结束演示")
                break
            continue

        idle_rounds = 0
        msg = json.loads(body.decode("utf-8"))
        print(f"[已收到] {msg}")
        channel.basic_ack(delivery_tag=method_frame.delivery_tag)

    channel.cancel()
    connection.close()

# 终端 2（生产者）
[已发送] {'id': 0, 'content': '第 0 条消息'}
[已发送] {'id': 1, 'content': '第 1 条消息'}
[已发送] {'id': 2, 'content': '第 2 条消息'}

# 终端 1（消费者）
[消费者] 等待消息中... 按 Ctrl+C 退出
[已收到] {'id': 0, 'content': '第 0 条消息'}
[已收到] {'id': 1, 'content': '第 1 条消息'}
[已收到] {'id': 2, 'content': '第 2 条消息'}