一、业务背景
在跨境电商订单系统中，不同模块之间需要异步通信。RabbitMQ提供了四种交换器（Exchange）类型，每种适用于不同的场景。
二、Direct Exchange：精准路由
Direct Exchange根据Routing Key精确匹配队列。
python
import pikaclass DirectExchangeExample: def init(self): self.connection = pika.BlockingConnection( pika.ConnectionParameters('localhost') ) self.channel = self.connection.channel() # 声明Direct Exchange self.channel.exchange_declare( exchange='order_direct', exchange_type='direct' ) def setup_queues(self): # 不同优先级的订单走不同队列 self.channel.queue_declare('order_high_priority') self.channel.queue_declare('order_normal_priority') self.channel.queue_declare('order_low_priority') # 绑定 self.channel.queue_bind( exchange='order_direct', queue='order_high_priority', routing_key='high' ) self.channel.queue_bind( exchange='order_direct', queue='order_normal_priority', routing_key='normal' ) self.channel.queue_bind( exchange='order_direct', queue='order_low_priority', routing_key='low' ) def publish_order(self, order, priority='normal'): self.channel.basic_publish( exchange='order_direct', routing_key=priority, body=json.dumps(order) )
使用场景：不同优先级的订单（VIP用户、普通用户、批量订单）分开处理。
三、Topic Exchange：模糊匹配
Topic Exchange根据通配符匹配Routing Key。
python
class TopicExchangeExample: def init(self): self.channel.exchange_declare( exchange='order_topic', exchange_type='topic' ) def setup_queues(self): # 日本订单队列 self.channel.queue_declare('order_jp') self.channel.queue_bind( exchange='order_topic', queue='order_jp', routing_key='order.jp.' ) # 美国订单队列 self.channel.queue_declare('order_us') self.channel.queue_bind( exchange='order_topic', queue='order_us', routing_key='order.us.' ) # 所有订单的日志队列 self.channel.queue_declare('order_log') self.channel.queue_bind( exchange='order_topic', queue='order_log', routing_key='order.#' # #匹配任意多个词 )
使用场景：按地域（国家、地区）路由订单，同时有一个全局日志队列。
四、Fanout Exchange：广播
Fanout Exchange把消息广播给所有绑定的队列。
python
class FanoutExchangeExample: def init(self): self.channel.exchange_declare( exchange='order_fanout', exchange_type='fanout' ) def setup_queues(self): # 所有消费者都收到同样的消息 self.channel.queue_declare('order_storage') # 存储服务 self.channel.queue_declare('order_notify') # 通知服务 self.channel.queue_declare('order_analytics') # 分析服务 self.channel.queue_bind('order_storage', 'order_fanout') self.channel.queue_bind('order_notify', 'order_fanout') self.channel.queue_bind('order_analytics', 'order_fanout') def publish_order(self, order): # 所有绑定的队列都会收到 self.channel.basic_publish( exchange='order_fanout', routing_key='', # Fanout忽略routing_key body=json.dumps(order) )
使用场景：订单状态变更需要通知多个服务（存储、通知、分析、日志）。
五、Headers Exchange：复杂条件匹配
Headers Exchange根据消息头中的键值对匹配，支持更复杂的条件。
python
class HeadersExchangeExample: def init(self): self.channel.exchange_declare( exchange='order_headers', exchange_type='headers' ) def setup_queues(self): # 大额订单队列 self.channel.queue_declare('order_large') self.channel.queue_bind( exchange='order_headers', queue='order_large', arguments={ 'x-match': 'all', # 所有条件都满足 'amount': 'large', 'status': 'paid' } ) # 紧急订单队列 self.channel.queue_declare('order_urgent') self.channel.queue_bind( exchange='order_headers', queue='order_urgent', arguments={ 'x-match': 'any', # 任一条件满足 'urgent': 'true', 'priority': 'high' } )
六、总结
四种交换器的选择原则：Direct用于精准路由、Topic用于模糊匹配、Fanout用于广播、Headers用于复杂条件。在跨境电商订单系统中，Direct用得最多（按订单类型路由），Fanout用于状态广播。
