Python实现RabbitMQ三种Exchange模式详解
本文通过Python Pika库详细演示了RabbitMQ中Fanout、Direct和Topic三种Exchange模式的具体实现。内容涵盖广播、精确路由及通配符路由的配置方法、代码示例及运行结果分析，帮助开发者理解不同场景下的消息分发机制。
近期在探索使用Redis构建消息队列的过程中，同时也调研了基于RabbitMQ的消息队列实现方案。本文主要总结并展示了RabbitMQ中三种核心Exchange模式的实现方式，分别为Fanout、Direct以及Topic模式。
基础连接配置代码base.py：
import pika
# 获取认证对象，参数是用户名、密码。远程连接时需要认证
credentials = pika.PlainCredentials("admin", "admin")
# BlockingConnection(): 实例化连接对象
# ConnectionParameters(): 实例化链接参数对象
connection = pika.BlockingConnection(pika.ConnectionParameters(
"192.168.0.102", 5672, "/", credentials))
# 创建新的channel(通道)
channel = connection.channel()
Fanout模式：该模式下，消息会被发送到所有绑定到指定Exchange的Queue中，消费者从各自的Queue中获取数据。这种机制类似于广播或发布/订阅模式。
相关代码实现如下：
发布者端代码publisher.py：
from base import channel, connection # 声明exchange, 不声明queue channel.exchange_declare(exchange="logs", exchange_type="fanout") # 广播 message = "hello fanout" channel.basic_publish( exchange="logs", routing_key="", body=message ) connection.close()
消费者端代码consumer.py：
from base import channel, connection
# 声明exchange
channel.exchange_declare(exchange="logs", exchange_type="fanout")
# 不指定queue名字, rabbitmq会随机分配一个名字, 消息处理完成后queue会自动删除
result = channel.queue_declare(exclusive=True)
# 获取queue名字
queue_name = result.method.queue
# 绑定exchange和queue
channel.queue_bind(exchange="logs", queue=queue_name)
def callback(ch, method, properties, body):
print("body:%s" % body)
channel.basic_consume(
callback,
queue=queue_name
)
channel.start_consuming()
Direct模式：发送端在发布消息时绑定一个routing_key1，而Queue在绑定时可以指定若干个routing_key2。当routing_key1与某个routing_key2完全相等，或者满足特定匹配规则时，消息将被路由至对应的Queue，并由相应的消费者从Queue中取出处理。
发布者端代码publisher.py：
from base import channel, connection channel.exchange_declare(exchange="direct_test", exchange_type="direct") message = "hello" channel.basic_publish( exchange="direct_test", routing_key="info", # 绑定key body=message ) connection.close()
消费者01代码consumer01.py：
from base import channel, connection
channel.exchange_declare(exchange="direct_test", exchange_type="direct")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
channel.queue_bind(
exchange="direct_test",
queue=queue_name,
# 绑定的key,与publisher中的相同
routing_key="info"
)
def callback(ch, method, properties, body):
print("body:%s" % body)
channel.basic_consume(
callback,
queue=queue_name
)
channel.start_consuming()
消费者02代码consumer02.py：
from base import channel, connection
channel.exchange_declare(exchange="direct_test", exchange_type="direct")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
channel.queue_bind(
exchange="direct_test",
queue=queue_name,
# 绑定的key
routing_key="error"
)
def callback(ch, method, properties, bosy):
print("body:%s" % body)
channel.basic_consume(
callback,
queue=queue_name
)
channel.start_consuming()
消费者03代码consumer03.py：
from base import channel, connection
channel.exchange_declare(exchange="direct_test", exchange_type="direct")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
key_list = ["info", "warning"]
for key in key_list:
channel.queue_bind(
exchange="direct_test",
queue=queue_name,
# 一个queue同时绑定多个key，有一个key满足条件时就可以收到数据
routing_key=key
)
def callback(ch, method, properties, body):
print("body:%s" % body)
channel.basic_consume(
callback,
queue=queue_name
)
channel.start_consuming()
执行测试命令：
python producer.py python consumer01.py python consumer02.py python consumer03.py
运行结果观察：
consumer01.py接收到的数据: body:b'hello' consumer02.py未接收到任何消息 consumer03.py接收到的数据: body:b'hello'
Topic模式相对复杂，以下是对其机制的理解：
在Topic模式中，发送端绑定的routing_key1与Queue绑定的若干个routing_key2进行匹配。只要routing_key1符合任意一个routing_key2所定义的模式规则，消息就会通过Exchange路由到该Queue，随后由接收端从Queue中取出。这可以看作是Direct模式的扩展，支持更灵活的路由规则。
具体的绑定方式如下：
发送端绑定配置：
channel.basic_publish(
exchange="topic_logs",
routing_key=routing_key,
body=message
)
接收端绑定配置：
channel.queue_bind(
exchange="topic_logs",
queue=queue_name,
routing_key=binding_key
)
发布者端代码publisher.py：
import sys from base import channel, connection # 声明exchange channel.exchange_declare(exchange="topic_test", exchange_type="topic") # 待发送消息 message = " ".join(sys.argv[1:]) or "hello topic" # 发布消息 channel.basic_publish( exchange="topic_test", routing_key="mysql.error", # 绑定的routing_key body=message ) connection.close()
消费者01代码consumer01.py：
from base import channel, connection
channel.exchange_declare(exchange="topic_test", exchange_type="topic")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
channel.queue_bind(
exchange="topic_test",
queue=queue_name,
routing_key="*.error" # 绑定的routing_key
)
def callback(ch, method, properties, body):
print("body:%s" % body)
channel.basic_consume(
callback,
queue=queue_name,
no_ack=True
)
channel.start_consuming()
消费者02代码consumer02.py：
from base import channel, connection
channel.exchange_declare(exchange="topic_test", exchange_type="topic")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
channel.queue_bind(
exchange="topic_test",
queue=queue_name,
routing_key="mysql.*" # 绑定的routing_key
)
def callback(ch, method, properties, body):
print("body:%s" % body)
channel.basic_consume(
callback,
queue=queue_name,
no_ack=True
)
channel.start_consuming()
执行测试命令：
python publisher02.py "this is a topic test" python consumer01.py python consumer02.py
运行结果观察：
consumer01.py接收到的数据: body:b'this is a topic test' consumer02.py接收到的数据: body:b'this is a topic test'
上述结果表明，通过配置相应的routing_key，两个消费者均成功接收到了消息。
接下来，将publisher.py中的routing_key修改为"mysql.info"。
再次执行测试命令：
python publisher02.py "this is a topic test" python consumer01.py python consumer02.py
运行结果观察：
consumer01.py未接收到任何消息 consumer02.py接收到的数据: body:b'this is a topic test'
通过这一对比示例，可以清晰地理解Topic模式的运行逻辑。
参考来源：https://www.jb51.net/article/150386.htm
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
