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

生产者_procudure_send_消息持久化
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
    'localhost'))  # rabbit默认端口5672 建立一个基本的 socket连接
channel = connection.channel()  # 声明一个管道 在管道里面发消息

# 声明queue
channel.queue_declare(queue='hello',durable=True)#durable=True队列持久化

# n RabbitMQ a message can never be sent directly to the queue, it always needs to go through an exchange.
channel.basic_publish(exchange='',
                      routing_key='hello',  # queue名字
                      body='Hello World!',  # body 发送的消息
                      properties=pika.BasicProperties(delivery_mode=2),#消息持久化
                          )
print(" [x] Sent 'Hello World!'")
connection.close()

消费者_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()