rabbitmq_消息持久化_消息公平分發(fā)_消息廣播

生產者_procudure_send_消息持久化
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
    'localhost'))  # rabbit默認端口5672 建立一個基本的 socket連接
channel = connection.channel()  # 聲明一個管道 在管道里面發(fā)消息

# 聲明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 發(fā)送的消息
                      properties=pika.BasicProperties(delivery_mode=2),#消息持久化
                          )
print(" [x] Sent 'Hello World!'")
connection.close()

消費者_consumer_recive_消息公平分發(fā)
 # _*_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:發(fā)給queue的信息
     print(" [x] Received %r" % body)
     ch.basic_ack(delivery_tag=method.delivery_tag)# 手動確認消息是否接收
 
 channel.basic_qos(prefetch_count=1)#根據(jù)權重發(fā)消息 一對一發(fā)
 channel.basic_consume(#消費消息
            callback,#如果收到消息，就調用CALLBACK函數(shù)來處理消息
            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)#綁定轉發(fā)器
 
 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()