生产者_procudure_send_消息持久化
importpika

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()

成都地区优秀IDC服务器托管提供商(成都创新互联).为客户提供专业的德阳机房服务器托管,四川各地服务器托管,德阳机房服务器托管、多线服务器托管.托管咨询专线：18982081108

 

消费者_consumer_recive_消息公平分发
# _*_coding:utf-8_*_
__author__ ='Alex Li'
importpika

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消费持久化的消息


defcallback(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_广播模式
importpika
importsys

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'
importpika

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')


defcallback(ch,method,properties, body):
   print(" [x] %r"% body)


channel.basic_consume(callback,
                     queue=queue_name,
                     no_ack=True)

channel.start_consuming()

分享文章：rabbitmq_消息持久化_消息公平分发_消息广播
转载源于：http://scyanting.com/article/gissii.html