4. RabbitMQ工作模式（二）：发布/订阅、路由与主题模式

上一章我们聊了简单队列和工作队列，说白了就是点对点的通信。但实际项目中，很多时候我们需要一条消息被多个消费者同时处理。比如订单创建后，既要发短信通知，又要更新库存，还要记录日志。这时候，发布/订阅模式就派上用场了。

我个人习惯把RabbitMQ的Exchange（交换机）比作一个邮局的分拣中心。生产者把消息扔给交换机，交换机根据规则分发给不同的队列。今天要讲的三种模式，其实就是交换机的三种路由策略。

4.1 发布/订阅模式（Publish/Subscribe）

这个模式的核心是Fanout交换机。它做的事情很简单——把收到的消息广播给所有绑定的队列。就像广播电台，谁在听都能收到。

我在项目中遇到过这样一个场景：用户下单后，需要同时触发邮件通知、短信通知和内部工单系统。如果用点对点的方式，得写三个生产者，太蠢了。用Fanout交换机，一条消息出去，三个队列各取所需。

来看代码实现：

import pika
import json

# 生产者：发送订单消息
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明Fanout交换机
channel.exchange_declare(
    exchange='order_fanout',
    exchange_type='fanout'
)

order = {
    'order_id': '2024001',
    'user_id': 'u1001',
    'amount': 299.00
}

channel.basic_publish(
    exchange='order_fanout',
    routing_key='',  # Fanout模式下routing_key被忽略
    body=json.dumps(order)
)

print(" [x] 订单消息已发送")
connection.close()

消费者这边，每个服务创建自己的临时队列，绑定到同一个交换机：

# 消费者：短信服务
def sms_callback(ch, method, properties, body):
    order = json.loads(body)
    print(f" [短信服务] 发送短信给用户 {order['user_id']}")
    print(f"         订单金额: {order['amount']}")

channel.exchange_declare(
    exchange='order_fanout',
    exchange_type='fanout'
)

# 创建临时队列，断开连接时自动删除
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

# 绑定到交换机
channel.queue_bind(
    exchange='order_fanout',
    queue=queue_name
)

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

channel.start_consuming()

4.2 Routing（路由）模式

发布/订阅模式虽然好用，但有个问题——它不管消息内容，一股脑全广播。实际业务中，我们经常需要根据消息的严重程度做分级处理。比如日志系统：error级别的日志要写入数据库并报警，info级别的日志只写入文件。

这时候就需要Direct交换机了。它根据routing_key精确匹配，把消息路由到对应的队列。

你想想看，这就像快递分拣——每个包裹上贴了标签（routing_key），分拣员看到标签就扔到对应的货架（队列）。

日志级别	routing_key	处理方式
ERROR	log.error	写入数据库 + 发送报警
WARNING	log.warning	写入文件 + 发送通知
INFO	log.info	仅写入文件

代码实现如下：

# 生产者：发送不同级别的日志
channel.exchange_declare(
    exchange='logs_direct',
    exchange_type='direct'
)

severities = ['error', 'warning', 'info']
for severity in severities:
    message = f"这是一条{severity}级别的日志"
    channel.basic_publish(
        exchange='logs_direct',
        routing_key=f'log.{severity}',
        body=message
    )
    print(f" [x] 发送 {severity}: {message}")

消费者可以灵活绑定自己关心的路由键：

# 消费者：只处理error和warning
channel.exchange_declare(
    exchange='logs_direct',
    exchange_type='direct'
)

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

# 绑定多个routing_key
channel.queue_bind(
    exchange='logs_direct',
    queue=queue_name,
    routing_key='log.error'
)
channel.queue_bind(
    exchange='logs_direct',
    queue=queue_name,
    routing_key='log.warning'
)

print(" [*] 等待日志消息...")

4.3 Topics（主题）模式

Direct模式虽然解决了分级问题，但还不够灵活。比如我想匹配所有以"log."开头的消息，或者匹配"log.error"和"log.warning"但不要"log.info"。这时候就需要Topic交换机了。

Topic模式支持通配符匹配：

• * 匹配一个单词
• # 匹配零个或多个单词

举个例子：

• log.* 匹配 log.error、log.info，但不匹配 log.error.db
• log.# 匹配 log.error、log.error.db、log.info.user
• *.error 匹配 log.error、db.error

我个人习惯用Topic模式来处理多维度分类的消息。比如一个电商系统，消息可以按"业务类型.操作.地域"来组织：

# 生产者：发送多维度消息
channel.exchange_declare(
    exchange='topic_orders',
    exchange_type='topic'
)

messages = [
    ('order.created.beijing', '北京用户下单'),
    ('order.paid.shanghai', '上海用户支付成功'),
    ('order.refund.guangzhou', '广州用户申请退款'),
    ('user.registered', '新用户注册')
]

for routing_key, message in messages:
    channel.basic_publish(
        exchange='topic_orders',
        routing_key=routing_key,
        body=message
    )
    print(f" [x] 发送 {routing_key}: {message}")

消费者可以灵活订阅：

# 消费者：订阅所有订单相关消息
channel.queue_bind(
    exchange='topic_orders',
    queue=queue_name,
    routing_key='order.#'
)

# 另一个消费者：只关注北京的订单
channel.queue_bind(
    exchange='topic_orders',
    queue=another_queue,
    routing_key='order.*.beijing'
)

4.4 三种模式对比

最后做个总结，方便你快速选择：

模式	交换机类型	路由规则	适用场景
发布/订阅	Fanout	广播所有消息	事件通知、日志广播
路由	Direct	精确匹配routing_key	日志分级、命令分发
主题	Topic	通配符匹配	多维度消息路由

嗯，这里要注意一点：不要一上来就用Topic模式。如果业务场景简单，Fanout或Direct就够用了。过度设计反而会增加维护成本。我在早期项目中就吃过这个亏，明明只需要广播，非要用Topic，结果配置了一堆通配符，自己都看晕了。

下一章我们会聊消息确认和持久化，这是保证消息不丢的关键。到时候见。