消费者模型:
1 import pika,time 2 3 consumer = pika.BlockingConnection 4 (pika.ConnectionParameters('localhost'))#创建socket连接 5 channel = consumer.channel()#创建管道 6 channel.exchange_declare(exchange='logs',exchange_type = 'fanout') 7 result = channel.queue_declare(exclusive=True)#exclusive排他,唯一的,自动删除连接。 8 queue_name = result.method.queue#创建一个随机的队列名称 9 print('queue_name',queue_name)#打印该队列名称 10 11 channel.queue_bind(exchange='logs', 12 queue = queue_name) 13 def backcall(ch,method,properties,body):#参数body是发送过来的消息。 14 print(ch,method,properties) 15 16 print('[x] Received %r'%body) 17 ch.basic_ack(delivery_tag=method.delivery_tag) 18 19 channel.basic_consume(backcall,#回调函数。执行结束后立即执行另外一个函数返回给发送端是否执行完毕。 20 queue=queue_name, 21 #no_ack=True#不会告知服务端我是否收到消息。一般注释。 22 )#如果注释掉,对方没有收到消息的话不会将消息丢失,始终在队列里等待下次发送。 23 24 print('waiting for message To exit press CTRL+C') 25 channel.start_consuming()#启动后进入死循环。一直等待消息。
生产者模型:
1 import pika 2 connection = pika.BlockingConnection( 3 pika.ConnectionParameters('localhost'))#建立一个最基本的socket 4 chanel = connection.channel()#声明一个管道 5 6 chanel.exchange_declare(exchange='logs',exchange_type='fanout')#生成转换器名字为logs并定义类型为fanout 7 8 chanel.basic_publish(exchange='logs',#转换器ex名字 9 routing_key='', 10 body ='HELLO WORD!', 11 ) 12 print( '发出一个消息') 13 connection.close()#关闭