参考 https://www.cnblogs.com/shenh/p/10497244.html
模式一:fanout
- 不用指定routing_key
- 消息会发给所有的订阅者
- 需要先启动订阅者
# 发布者
import json
import pika
credentials = pika.PlainCredentials('root', 'root') # mq的用户名、密码
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='10.131.70.49', port=5672, # mq的IP、端口
virtual_host='/',
credentials=credentials
))
channel = connection.channel()
channel.exchange_declare(
exchange='python-text', # 指定队列
durable=True, # 消息持久化
exchange_type='fanout' # 发布的模式
)
for i in range(10):
message = json.dumps({"Order":"1000%s"%i})
channel.basic_publish(
exchange='python-text',
routing_key='',
body=message,
)
connection.close()
# 订阅者
import pika
credentials = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='10.131.70.49', port=5672, virtual_host='/', credentials=credentials))
channel = connection.channel()
# 创建临时队列,consumer关闭后,队列自动删除
result = channel.queue_declare('', exclusive=True)
# 声明exchange,由exchange指定消息在哪个队列传递,如不存在,则创建。durable = True 代表exchange持久化存储,False 非持久化存储
channel.exchange_declare(exchange='python-text', durable=True, exchange_type='fanout')
# 绑定exchange和队列 exchange 使我们能够确切地指定消息应该到哪个队列去
channel.queue_bind(exchange='python-text', queue=result.method.queue)
# 定义一个回调函数来处理消息队列中的消息,这里是打印出来
def callback(ch, method, properties, body):
ch.basic_ack(delivery_tag=method.delivery_tag)
print(body.decode())
channel.basic_consume(
result.method.queue,
callback,
# 设置成 False,在调用callback函数时,未收到确认标识,消息会重回队列。True,无论调用callback成功与否,消息都被消费掉
False)
channel.start_consuming()
模式二:direct
- 传递或接受消息时需要指定 routing_key
这里发布了两种,10个OrderID和一个OrderIDd
# 发布
import json
import pika
credentials = pika.PlainCredentials('root', 'root') # mq的用户名、密码
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='10.131.70.49', port=5672, # mq的IP、端口
virtual_host='/',
credentials=credentials
))
channel = connection.channel()
channel.exchange_declare(
exchange='python-text', # 指定队列
durable=True, # 消息持久化
exchange_type='direct' # 发布的模式
)
for i in range(10):
message = json.dumps({"Order": "1000%s" % i})
channel.basic_publish(
exchange='python-text',
routing_key='OrderID',
body=message,
)
message = json.dumps({"Order": "others"})
channel.basic_publish(
exchange='python-text',
routing_key='OrderIDd',
body=message,
)
connection.close()
这里订阅OrderID会接收到十条消息
# 订阅OrderID
import pika
credentials = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='10.131.70.49', port=5672, virtual_host='/', credentials=credentials))
channel = connection.channel()
# 创建临时队列,consumer关闭后,队列自动删除
result = channel.queue_declare('', exclusive=True)
# 声明exchange,由exchange指定消息在哪个队列传递,如不存在,则创建。durable = True 代表exchange持久化存储,False 非持久化存储
channel.exchange_declare(exchange='python-text', durable=True, exchange_type='direct')
# 绑定exchange和队列 exchange 使我们能够确切地指定消息应该到哪个队列去
channel.queue_bind(
exchange='python-text', queue=result.method.queue,routing_key='OrderID'
)
# 定义一个回调函数来处理消息队列中的消息,这里是打印出来
def callback(ch, method, properties, body):
ch.basic_ack(delivery_tag=method.delivery_tag)
print(body.decode())
channel.basic_consume(
result.method.queue,
callback,
# 设置成 False,在调用callback函数时,未收到确认标识,消息会重回队列。True,无论调用callback成功与否,消息都被消费掉
False)
channel.start_consuming()
这里订阅OrderIDd会接收到一条消息
# 订阅OrderIDd
import pika
credentials = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='10.131.70.49', port=5672, virtual_host='/', credentials=credentials))
channel = connection.channel()
# 创建临时队列,consumer关闭后,队列自动删除
result = channel.queue_declare('', exclusive=True)
# 声明exchange,由exchange指定消息在哪个队列传递,如不存在,则创建。durable = True 代表exchange持久化存储,False 非持久化存储
channel.exchange_declare(exchange='python-text', durable=True, exchange_type='direct')
# 绑定exchange和队列 exchange 使我们能够确切地指定消息应该到哪个队列去
channel.queue_bind(
exchange='python-text', queue=result.method.queue,routing_key='OrderIDd'
)
# 定义一个回调函数来处理消息队列中的消息,这里是打印出来
def callback(ch, method, properties, body):
ch.basic_ack(delivery_tag=method.delivery_tag)
print(body.decode())
channel.basic_consume(
result.method.queue,
callback,
# 设置成 False,在调用callback函数时,未收到确认标识,消息会重回队列。True,无论调用callback成功与否,消息都被消费掉
False)
channel.start_consuming()
网友评论