rabbitmq_消息持久化_消息公平分发_消息广播

消费者_consumer_recive_消息公平分发 # _*_coding:utf-8_*_ __author__ = 'Alex Li' import pika  connection = pika.BlockingConnection(pika.ConnectionParameters(     'localhost')) #rabbit默认端口5672 建立一个基本的 socket连接 channel = connection.channel()#声明一个管道 在管道里面收消息  # You may ask why we declare the queue again ‒ we have already declared it in our previous code. # We could avoid that if we were sure that the queue already exists. For example if send.py program # was run before. But we're not yet sure which program to run first. In such cases it's a good # practice to repeat declaring the queue in both programs. channel.queue_declare(queue='hello1')#声明queue 消费持久化的消息   def callback(ch, method, properties, body):#处理消息     print("---->",ch,method,properties)#ch 管道内存对象地址 method:发给queue的信息     print(" [x] Received %r" % body)     ch.basic_ack(delivery_tag=method.delivery_tag)# 手动确认消息是否接收  channel.basic_qos(prefetch_count=1)#根据权重发消息 一对一发 channel.basic_consume(#消费消息            callback,#如果收到消息，就调用CALLBACK函数来处理消息            queue='hello1',#从哪个队列里收消息            #no_ack=True,             )  print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming()#启动 开始收消息 一直收，没有就卡主

生产者_send_广播模式 import pika import sys  connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel()  channel.exchange_declare(exchange='logging',exchange_type='fanout')  messages = ' '.join(sys.argv[1:]) or "info: Hello World!" #messages="info: Hello World!" channel.basic_publish(exchange='yyyy',                       routing_key='',                       body=messages) print(" [x] Sent %r" % messages) connection.close()

消费者_recive_广播模式 # _*_coding:utf-8_*_ __author__ = 'Alex Li' import pika  connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel()  channel.exchange_declare(exchange='yyyy',                          exchange_type='fanout')  result = channel.queue_declare(exclusive=True)  # 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除 queue_name = result.method.queue # 随机分配的queue print("random queue name:",queue_name)  channel.queue_bind(exchange='yyyy',                    queue=queue_name)#绑定转发器  print(' [*] Waiting for logs. To exit press CTRL+C')   def callback(ch, method, properties, body):     print(" [x] %r" % body)   channel.basic_consume(callback,                       queue=queue_name,                       no_ack=True)  channel.start_consuming()