rabbitmq-任务轮询队列

工作队列

我们将创建一个工作队列（Work Queue），它会发送一些耗时的任务给多个工作者（Worker）。

工作队列（又称：任务队列——Task Queues）是为了避免等待一些占用大量资源、时间的操作。当我们把任务（Task）当作消息发送到队列中，一个运行在后台的工作者（worker）进程就会取出任务然后处理。当你运行多个工作者（workers），任务就会在它们之间共享。

循环调度:

使用工作队列的一个好处就是它能够并行的处理队列。如果堆积了很多任务，我们只需要添加更多的工作者（workers）就可以了，扩展很简单。

RabbitMQ会按顺序得把消息发送给每个消费者（consumer）。平均每个消费者都会收到同等数量得消息。这种发送消息得方式叫做——轮询（round-robin）。

'''
Created on 2016年12月13日

@author: test
'''

'''
基于队列的消息发送,轮寻每个队列监听者
生产者，在投递消息时必须指定消息投递的队列
持久化声明,有助于rabbit崩溃时消息恢复
@1 队列持久化
@2 消息持久化
'''

import pika
import uuid

if __name__ == '__main__':
username = 'test'   #指定远程rabbitmq的用户名密码
pwd = 'test'
user_pwd = pika.PlainCredentials(username, pwd)
#链接rabbitmq
connection = pika.BlockingConnection(pika.ConnectionParameters(host="192.168.1.109",virtual_host="test_vhost",credentials=user_pwd))

#申请信道
channel = connection.channel()
#定义队列
channel.queue_declare(queue="task_queue", #队列名字
durable=False)#声明队列持久化,已经声明过非持久化的队列不能再声明为持久化队列:True

uid = str(uuid.uuid4())
#发布消息
channel.basic_publish(exchange="",#定义匿名交换机,消息会根据routing_key发到指定队列上
routing_key="task_queue", #指定路由的队列
body="{'taskuuid': 1, 'siteid': [1], 'type': 'data', 'data': {'id':200, 'url':'asdasd'}}",
properties=pika.BasicProperties(
#delivery_mode=2,#声明消息持久化,把消息存放硬盘
correlation_id=uid
))
#关闭链接
connection.close()

'''
Created on 2016年12月14日

@author: test
'''

'''
基于队列的消息发送
消费者，只需在指定的队列上监听消息即可
建立消息回复机制，可以有效的避免消息丢失
'''

import pika

if __name__ == '__main__':
username = 'test'   #指定远程rabbitmq的用户名密码
pwd = 'test'
user_pwd = pika.PlainCredentials(username, pwd)
#链接rabbitmq
connection = pika.BlockingConnection(pika.ConnectionParameters(host="192.168.1.109",virtual_host="test_vhost",credentials=user_pwd))

#申请信道
channel = connection.channel()

#声明队列，避免异常
channel.queue_declare(queue="task_queue",durable=False)

#消息处理函数
def callback(ch,method,properties,body):
#处理一些耗时的操作
print(body,properties.correlation_id)
#回复消息发布者，消息处理完毕
ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)#声明每个消费者某时刻只能接一个任务

#注册
channel.basic_consume(callback,
queue="task_queue",
no_ack=False) #开启或者关闭消息处理回复，建议开启回复：False

channel.start_consuming()