在下面的Python程序中,我正在配置RabbitMq。我正在创建一个名为“order”的交换,并发布3条消息,其中包含路由键“order.notify”,“order.report”,“order.test”。
import pika
import json
import uuid
con = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = con.channel()
channel.exchange_declare(
exchange='order',
exchange_type = 'direct'
)
channel.basic_publish(
exchange= 'order',
routing_key= 'order.notify',
body= json.dumps({'user_email' : 'First'})
#body= json.dumps({'user_email' : order['user_email']})
)
print('[x] Sent nortify message')
channel.basic_publish(
exchange= 'order',
routing_key= 'order.report',
body= json.dumps({'user_email' : 'Second'})
)
print('[x] Sent report message')
channel.basic_publish(
exchange= 'order',
routing_key= 'order.test',
body= json.dumps({'user_email' : 'third'})
)
print('[x] Sent report message')
con.close()
现在,在消费者端,我只创建了2个队列,绑定键为order.nortiy和order.report
report.py
import pika
import json
con = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = con.channel()
queue = channel.queue_declare('order_notify')
queue_name = queue.method.queue
channel.queue_bind(
exchange='order',
queue=queue_name,
routing_key='order.report' #binding key
)
def callback(ch,method, properties, body):
payload = json.loads(body)
# print(' [x] Notifying {}' .format(payload['user_email']))
print('Report Queue')
print(payload['user_email'])
ch.basic_ack(delivery_tag= method.delivery_tag)
channel.basic_consume(on_message_callback= callback,queue=queue_name)
print(' [*] waiting for report messages. To exit press CTRL + C')
channel.start_consuming()
nortify.py
import pika
import json
con = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = con.channel()
queue = channel.queue_declare('order_notify')
queue_name = queue.method.queue
channel.queue_bind(
exchange='order',
queue=queue_name,
routing_key='order.notify' #binding key
)
def callback(ch,method, properties, body):
payload = json.loads(body)
# print(' [x] Notifying {}' .format(payload['user_email']))
print('Nortify Queue')
print(payload['user_email'])
ch.basic_ack(delivery_tag= method.delivery_tag)
channel.basic_consume(on_message_callback= callback,queue=queue_name)
print(' [*] waiting for report messages. To exit press CTRL + C')
channel.start_consuming()
现在哪个队列将消耗哪个消息。我试着运行,队列随机消耗。有人能解释一下吗?
尝试运行上面的程序,但得到的结果是随机的。每个队列在运行不同的时间消耗不同的消息
1条答案
按热度按时间mhd8tkvw1#
您有两个队列:
1.具有路由键的队列:“order.notify”应该获取所有使用“order.notify”路由键生成的消息。
1.具有路由键的队列:“order.report”应获取使用“order.report”路由键生成的所有消息。
最后一条带有“order.test”路由键的消息将丢失,因为没有队列使用它。