#coding:utf-8
import pika
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd)) #建立一个实例
channel = connection.channel() # 声明一个管道，在管道里发消息
channel.queue_declare(queue=‘hello‘) #在管道里声明queue，并指定queue_name
#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! ‘) # 消息内容
print(" [x] Sent ‘Hello World!‘")
connection.close() # 队列关闭接收端：
#coding:utf-8
import pika
import time
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.206‘, credentials=user_pwd)) # 建立实例
channel = connection.channel() # 声明管道
# 在不确定procedure与consumer哪个先运行的情况下，在consumer端 再次声明QUEUE。
channel.queue_declare(queue=‘hello‘)
def callback(ch, method, properties, body): #定义回调函数，四个参数为标准格式
print(" [x] Received %r" % body) # 输出接收到的消息
time.sleep(15)
ch.basic_ack(delivery_tag=method.delivery_tag) # 告诉生产者消息处理完成
channel.basic_consume( # 消费消息
callback, # 如果收到消息，就调用callback函数来处理消息
queue=‘hello‘, # 你要从那个队列里收消息
#no_ack=True # True 表示consumer取完消息后不给procedure发送ack确认，注意本例中如果为True，它会与callback函数中的basic_ack语句产生冲突；默认False
)
print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
channel.start_consuming() # 开始消费消息验证：
在多个shell窗口中运行consumer，然后运行procedure，发现每个consumer 轮流接收相同队列中的消息。
二、
发送端：
#coding:utf-8
import pika
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
channel = connection.channel()
channel.queue_declare(queue=‘hello2‘, durable=True) #创建一个新队列task_queue，设置队列持久化，注意不要跟已存在的队列重名，否则有报错
#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=‘hello2‘, body=‘Hello World 2 !‘,
properties=pika.BasicProperties(delivery_mode=2, ) # make message persistent
)
print(" [x] Sent ‘Hello World!‘")
connection.close()接收端：
#coding:utf-8
import pika
import time
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
channel = connection.channel()
channel.queue_declare(queue=‘hello2‘, durable=True)
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(10)
ch.basic_ack(delivery_tag=method.delivery_tag) # 告诉生产者，消息处理完成
channel.basic_qos(prefetch_count=1) # 类似权重，按能力分发，如果有一个消息，就不在给你发
channel.basic_consume( # 消费消息
callback, # 如果收到消息，就调用callback
queue=‘hello2‘,
)
print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
channel.start_consuming()(1) rabbitmq循环调度，将消息循环发送给不同的消费者
(2) 消息确认机制，为了确保一个消息不会丢失，RabbitMQ支持消息的确认 , 一个 ack(acknowlegement) 是从消费者端发送一个确认去告诉RabbitMQ 消息已经接收了、处理了，RabbitMQ可以释放并删除掉了。如果一个消费者死掉了（channel关闭、connection关闭、或者TCP连接断开了）而没有发送ack，RabbitMQ 就会认为这个消息没有被消费者处理，并会重新发送到生产者的队列里，如果同时有另外一个消费者在线，rabbitmq将会将消息很快转发到另外一个消费者中。
(3) 消息持久化，将消息写入硬盘中。RabbitMQ不允许你重新定义一个已经存在、但属性不同的queue。关于消息为持久化， 需要生产者定义持久化的queue，以及通过设置 delivery_mode 属性为 2来标记本消息持久化。
(4) 公平调度。在一个消费者未处理完一个消息之前不分发新的消息给它，而是将这个新消息分发给另一个不是很忙的消费者进行处理。为了解决这个问题可以在消费者代码中使用 channel.basic.qos ( prefetch_count = 1 )，将消费者设置为公平调度
验证：
在多个shell窗口中打开consumer，然后运行procedure，观察收到消息为RR
三、
exchange：交换机。生产者不是将消息发送给队列，而是将消息发送给交换机，由交换机决定将消息发送给哪个队列。所以exchange必须准确知道消息是要送到哪个队列，还是要被丢弃。因此要在exchange中给exchange定义规则，所有的规则都是在exchange的类型中定义的。
(1) fanout
生产者：
#coding:utf-8
import pika
import sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
#建立一个实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
channel = connection.channel()
# 注意：这里是广播，不需要声明queue
channel.exchange_declare(‘logs‘, ‘fanout‘)
message = ‘ ‘.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange=‘logs‘,
routing_key=‘‘, # 注意此处空，必须有
body=message)
print(" [x] Sent %r" % message)
connection.close()消费者：
#coding:utf-8
import pika
import time
user_pwd = pika.PlainCredentials(‘admin‘,‘admin‘)
# 建立实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
# 声明管道
channel = connection.channel()
channel.exchange_declare(‘logs‘, ‘fanout‘)
# 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
result = channel.queue_declare(exclusive=True)
# 获取随机的queue名字
queue_name = result.method.queue
#print("random queuename:", queue_name)
channel.queue_bind(exchange=‘logs‘, # queue绑定到转发器上
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()
# 注意：广播是实时的，收不到就没了，消息不会存下来，类似收音机
(2) direct
生产者：
#coding:utf-8
import pika
import sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
#建立一个实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
channel = connection.channel()
channel.exchange_declare(‘direct_logs‘, ‘direct‘)
# 重要程度级别，这里默认定义为 info
severity = sys.argv[1] if len(sys.argv) > 1 else ‘warning‘
message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World! direct log‘
channel.basic_publish(exchange=‘direct_logs‘, routing_key=severity, body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()消费者：
#coding:utf-8
import pika
import time, sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
# 建立实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
channel = connection.channel()
channel.exchange_declare(‘direct_logs‘, ‘direct‘)
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
# 获取运行脚本所有的参数
severities = sys.argv[1:]
if not severities:
#severities = [‘error‘, ‘warning‘]
sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
sys.exit(1)
# 循环列表去绑定
for severity in severities:
channel.queue_bind(exchange=‘direct_logs‘, queue=queue_name, routing_key=severity)
print(‘ [*] Waiting for logs. To exit press CTRL+C‘)
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()
(3) topic
生产者:
#coding:utf-8
import pika
import sys
user_pwd = pika.PlainCredentials(‘admin‘, ‘admin‘)
#建立一个实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, 5672, credentials=user_pwd))
# 声明一个管道，在管道里发消息
channel = connection.channel()
channel.exchange_declare(‘topic_logs‘, ‘topic‘)
routing_key = sys.argv[1] if len(sys.argv) > 1 else ‘anonymous.info‘
message = ‘ ‘.join(sys.argv[2:]) or ‘Hello World!‘
channel.basic_publish(exchange=‘topic_logs‘, routing_key=routing_key, body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()
消费者：
#coding:utf-8
import pika
import time, sys
user_pwd = pika.PlainCredentials(‘admin‘,‘admin‘)
# 建立实例
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.40.119‘, credentials=user_pwd))
channel = connection.channel()
channel.exchange_declare(‘topic_logs‘, ‘topic‘)
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
binding_keys = sys.argv[1:]
if not binding_keys:
sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
sys.exit(1)
for binding_key in binding_keys:
channel.queue_bind(exchange=‘topic_logs‘, queue=queue_name, routing_key=binding_key)
print(‘ [*] Waiting for logs. To exit press CTRL+C‘)
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()
原文地址：http://blog.51cto.com/caiyuanji/2122527
