OnMyHeart 2019-12-07
生产者:
# !/usr/bin/env python # -*- coding: utf-8 -*- import pika # 创建连接对象 connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘)) channel = connection.channel() # 创建交换机 channel.exchange_declare(exchange=‘logs‘, exchange_type=‘fanout‘) # 往队列里插入数据 channel.basic_publish(exchange=‘logs‘, routing_key=‘‘, body="I don‘t know") connection.close()
消费者:
# !/usr/bin/env python # -*- coding: utf-8 -*- import pika # 创建连接对象 connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘)) channel = connection.channel() # 创建交换机 channel.exchange_declare(exchange=‘logs‘, exchange_type=‘fanout‘) # 创建随机队列 result = channel.queue_declare(queue=‘s3‘) # queue_name = result.method.queue # 随机队列名 # 将队列绑定到指定的交换机上 channel.queue_bind(exchange=‘logs‘, queue=‘s3‘) def callback(ch, method, properties, body): print(" [x] Received %s" % body) channel.basic_consume(queue=‘s3‘, on_message_callback=callback, auto_ack=True) # 无应答模式 channel.start_consuming()