这篇文章将会介绍关于RabbbitMQ的使用，并且使用的是kombo（客户端的Python实现）来实现；

如果使用的是mac安装的话，可以先安装到指定的位置，接着配置命令访问路径：

1. cd ~

2. vi .bash_profile，输入下面两行

   RABBIT_HOME=/usr/local/Cellar/rabbitmq/3.6.9_1
   
   PATH=$PATH:$RABBIT_HOME/sbin
   

3. esc，:wq保存并退出即可

开始：sudo rabbitmq-server start结束：sudo rabbitmq-server stop

首先我们需要知道Producer和Consumer的初始化和其对应的publish和consumer方法。

class kombu.Producer(channel, exchange=None, routing_key=None,

serializer=None, auto_declare=None,

compression=None, on_return=None)

# 发布消息

.publish(body, routing_key=None, delivery_mode=None, mandatory=False,

immediate=False, priority=0, content_type=None, content_encoding=None,

serializer=None, headers=None, compression=None, exchange=None,

retry=False, retry_policy=None, declare=None, expiration=None, **properties)

class kombu.Consumer(channel, queues=None, no_ack=None,

auto_declare=None, callbacks=None, on_decode_error=None,

on_message=None, accept=None, prefetch_count=None, tag_prefix=None)

# 消费

.consume(no_ack=None)

当收到消息的时候，除非你已经对这个message进行了相关的操作，否则像是某个消费者的通道关闭等特殊情况下，RabbitMQ不会丢失掉这个信息，如果存在其它的消费者，则丢给其它消费者，没有就扔回队列中；当然你也可以通过no_ack=True来关闭消息确认机制。

from kombu import Exchange, Queue, Connection, Consumer, Producer

task_queue = Queue('tasks', exchange=Exchange('tasks', type='direct'), routing_key='tasks')

# 生产者

with Connection('amqp://guest@localhost:5672//') as conn:

    with conn.channel() as channel:

        producer = Producer(channel)

        producer.publish({'hello': 'world'},

                         retry=True,

                         exchange=task_queue.exchange,

                         routing_key=task_queue.routing_key,

                         declare=[task_queue])

def get_message(body, message):

    print(body)

    # message.ack()

# 消费者

with Connection('amqp://guest@localhost:5672//') as conn:

    with conn.channel() as channel:

        consumer = Consumer(channel, queues=task_queue, callbacks=[get_message,], prefetch_count=10)

        consumer.consume(no_ack=True)

生产者和消费者相互对应，这样一个简易的消息队列就可以使用了。

我们将创建一个工作队列，专门用来处理分配耗时的任务。原理就是将任务封装成一个消息，由客户端发送到消息队列中，而后台运行的工作进程负责弹出任务并且分配给消费者来执行任务。这种方案在一些IO密集型的情况下很有用，比如在短时间内HTTP请求窗口中无法处理复杂的任务。

1. 我们先创建相关的exchange和queue，queues.py文件如下：

   from kombu import Exchange, Queue
   
   task_exchange = Exchange('tasks', type='direct')
   
   task_queues = [Queue('high', exchange=task_exchange, routing_key='high'),
   
                  Queue('middle', exchange=task_exchange, routing_key='middle'),
   
                  Queue('low', exchange=task_exchange, routing_key='low')]
   

2. 接下来再创建消费者，worker.py文件如下：

   from kombu.mixins import ConsumerMixin
   
   from queues import task_queues
   
   # 消费者
   
   class Worker(ConsumerMixin):
   
       def __init__(self, connection):
   
           self.connection = connection
   
       def get_consumers(self, Consumer, channel):
   
           consumer = Consumer(queues=task_queues, callbacks=[self.process_task], accept=['text/plain', 'json', 'pickle'])
   
           consumer.qos(prefetch_count=10)  # 最多一下子获取10个任务
   
           return [consumer]
   
       def process_task(self, body, message):
   
           fun = body['fun']; args = body['args']; kwargs = body['kwargs']
   
           try:
   
               fun(*args, **kwargs)
   
           except Exception as exc:
   
               print(exc)
   
               message.requeue()
   
           else:
   
               message.ack()
   
   if __name__ == '__main__':
   
       from kombu import Connection
   
       with Connection('amqp://guest@localhost:5672//') as conn:
   
           try:
   
               worker = Worker(conn)
   
               worker.run()
   
           except KeyboardInterrupt:
   
               print('bye bye')
   

3. 创建需要传递给消费者执行的任务，tasks.py如下：

   def hello_task(who='world'):
   
       import time
   
       print('wait one second')
   
       time.sleep(1)
   
       print('Hello {}'.format(who))
   

4. 最后，创建生产者，client.py如下：

   from kombu.pools import producers
   
   from queues import task_exchange
   
   routing_keys = {
   
       'high': 'high',
   
       'middle': 'middle',
   
       'low': 'low'
   
   }
   
   # 将消息序列化后发送到队列中
   
   def send_as_task(connection, fun, key='middle', args=(), kwargs={}):
   
       payload = {'fun': fun, 'args': args, 'kwargs': kwargs}
   
       routing_key = routing_keys[key]
   
       with producers[connection].acquire(block=True) as producer:
   
           producer.publish(payload, serializer='pickle', exchange=task_exchange,
   
                            routing_key=routing_key, declare=[task_exchange])
   
   if __name__ == '__main__':
   
       from kombu import Connection
   
       from tasks import hello_task
   
       with Connection('amqp://guest@localhost:5672//') as conn:
   
           send_as_task(conn, fun=hello_task, args=('wang',))
   

上面的代码主要实现的是，将hello_task这个任务经过pickle序列化以后发送到指定的middle消息队列中，接着消费者（可以开多个进程）从中取出消息后再执行任务。