#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@author: zengchunyun
"""
import pika class MQServer(object):
def __init__(self, host, port=5672, exchange=None, exchange_type="topic"):
"""
初始化MQ设置
:param host: MQ服务器地址
:param port: MQ端口
:param exchange: 交换器名称
:param exchange_type: 交换器类型,默认关键字类型
:return:
"""
self.host = host
self.port = port
self.exchange = exchange
self.exchange_type = exchange_type
self.queue = None
self.connection = self.connect()
self.channel = self.connect_channel()
self.create_exchange() def connect(self):
"""
连接MQ服务器
:return:
"""
return pika.BlockingConnection(pika.ConnectionParameters(host=self.host, port=self.port)) def connect_channel(self):
"""
创建频道
:return:
"""
return self.connection.channel() def create_exchange(self):
"""
定义交换器名称,防止发布时,如果交换器不存在,异常
:return:
"""
self.channel.exchange_declare(exchange=self.exchange, type=self.exchange_type) def publish(self, exchange=None, routing_key=None, body=None):
"""
创建发布者
:param exchange: 交换器名称
:param routing_key: 路由KEY
:param body:消息主体
:return:
"""
if exchange:
self.exchange = exchange
self.channel.basic_publish(exchange=self.exchange, routing_key=routing_key, body=body)
self.close() def consumer(self, exchange=None, routing_key=None, callback=None):
"""
创建消费者
:param exchange:
:param routing_key:
:param callback:
:return:
"""
if exchange:
self.exchange = exchange
self.create_queue()
self.channel.queue_bind(queue=self.queue, exchange=self.exchange, routing_key=routing_key)
self.channel.basic_consume(consumer_callback=callback, queue=self.queue, no_ack=True)
self.start() def create_queue(self):
"""
生成队列,当关闭consumer时,加上exclusive=True,queue也会被删除
:return:
"""
self.queue = self.channel.queue_declare(exclusive=True).method.queue # 为每个消费者生成不同的队列 def close(self):
"""
关闭消息连接
:return:
"""
self.connection.close() def start(self):
self.channel.start_consuming()
1.消息持久化存储
虽然有了消息反馈机制，但如果rabbitmq自身挂掉的话，那么任务还是会丢失，所以需要将任务持久化存储起来，
durable=True # 开启持久化设置，rabbitmq不允许使用不同的参数来重新定义存在的队列
self.queue = self.channel.queue_declare(exclusive=True,durable=True)
self.channel.exchange_declare(exchange=self.exchange, type=self.exchange_type, durable=True)
在发送任务的时候，用delivery_mode＝2来标记任务为持久化存储
self.channel.basic_publish(exchange='',
routing_key=routing_key,
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))
2.公平调度（fair dispatch)
虽然每个工作者是依次分配到任务，但是每个任务不一定一样，可能有到任务比较重，执行时间长，有的任务比较轻，执行时间短，如果能公平调度最好了，使用basic_qos设置prefetch_count=1，使得rabbitmq不会在同一时间给工作者分配多个任务，即只有工作者完成任务之后，才会再次接收到任务
channel.basic_qos(prefetch_count=1)
完整示例代码
#!/usr/bin/env python
import pika
import sys connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel() channel.queue_declare(queue='task_queue', durable=True) message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))
print (" [x] Sent %r" % (message,))
connection.close()
消费者代码
#!/usr/bin/env python
import pika
import time connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel() channel.queue_declare(queue='task_queue', durable=True)
print( ' [*] Waiting for messages. To exit press CTRL+C') def callback(ch, method, properties, body):
print (" [x] Received %r" % (body,))
time.sleep( body.count('.') )
print (" [x] Done")
ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
queue='task_queue') channel.start_consuming()