Python RabbitMQ 核心原理、模式与实战场景解析

消费者示例：

conn = pika.BlockingConnection(config)
channel = conn.channel()
channel.exchange_declare(exchange='ceshi', exchange_type='direct')
channel.queue_declare(queue='hello')
channel.queue_bind(exchange='ceshi', queue='hello', routing_key='1')

def callback(channel, method, properties, body):
    print(f'Received: {body}')
    channel.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(callback, queue='hello', no_ack=False)
print('Waiting...')
channel.start_consuming()

提示：修改 RoutingKey 后需重建队列以生效。

2. Fanout 扇形交换机

忽略 RoutingKey，将消息广播给所有绑定队列。适用于状态通知或群聊消息分发。

生产者：

channel.exchange_declare(exchange='ceshi2', exchange_type='fanout')
channel.basic_publish(exchange='ceshi2', routing_key='', body='Broadcast!')

消费者：

result = channel.queue_declare(exclusive=True) # 生成临时队列名
queue_name = result.method.queue
channel.queue_bind(exchange='ceshi2', queue=queue_name, routing_key='')

3. Topic 主题交换机

支持通配符匹配。* 匹配一个单词，# 匹配零个或多个单词。适合分类新闻或日志路由。

匹配规则：

• 生产者 routing_key='a.b.c.d'
• 消费者 binding_key='a.*' -> 不匹配（* 仅匹配一层）
• 消费者 binding_key='a.#' -> 匹配成功

4. RPC 远程过程调用

实现同步请求响应模式。生产者创建临时队列接收结果，携带 CorrelationID 确保响应匹配。

RPC 客户端：

import uuid
class RpcClient:
    def __init__(self):
        self.conn = pika.BlockingConnection(config)
        self.channel = self.conn.channel()
        result = self.channel.queue_declare(exclusive=True)
        self.queue_name = result.method.queue
        self.channel.basic_consume(self.on_request, queue=self.queue_name, no_ack=True)

    def on_request(self, channel, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, message):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(
            exchange='', routing_key='rpc',
            properties=pika.BasicProperties(reply_to=self.queue_name, correlation_id=self.corr_id),
            body=message
        )
        while self.response is None:
            self.conn.process_data_events()
        return self.response

RPC 服务端：

def on_request(channel, method, props, body):
    print(f'Processing: {body}')
    channel.basic_publish(
        exchange='', routing_key=props.reply_to,
        properties=pika.BasicProperties(correlation_id=props.correlation_id),
        body='Done'
    )
    channel.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue='rpc')

典型应用场景

1. 任务依赖解耦

传统 Crontab 难以应对动态任务依赖。使用 RabbitMQ 后，任务间通过消息触发。例如任务 2 等待任务 3 完成信号后再执行，无需硬编码时间间隔，系统更灵活。

2. 异步通知与解耦

用户发布文章后，系统只需关注发布动作，后续推荐、置顶等操作由后台服务异步处理。新增功能无需修改发布代码，降低耦合度。

3. 长耗时任务回调

用户发起耗时操作 A，服务立即返回'已受理'，后台执行完成后发送结果至 RabbitMQ，用户订阅该消息获取最终状态。避免阻塞主线程，提升用户体验。

管理后台启用

RabbitMQ 内置 Web 管理界面，安装后需手动开启插件：

rabbitmq-plugins enable rabbitmq_management

访问 http://localhost:15672/，默认账号密码均为 guest。