1、安装RabbitMQ

参考文档：

Windows 上下载安装 RabbitMQ 的方法步骤

Linux 上下载安装 RabbitMQ 的方法步骤

2、安装pika

pika 是 AMQP 0-9-1 协议的纯 Python 实现，它基本保持完全独立于底层网络支持库。Pika可以通过PyPI下载，并可以使用easy_install或pip安装:

pip install pika

或

easy_install pika

相关文档：https://pika.readthedocs.io/en/stable/

3、使用说明

1）关键词说明

2）消息队列运行机制

客户端连接到消息队列服务器，打开一个channel。
客户端声明一个exchange，并设置相关属性。
客户端声明一个queue，并设置相关属性。
客户端使用routing key，在exchange和queue之间建立好绑定关系。
客户端投递消息到exchange。

Exchange接收到消息后，就根据消息的key和已经设置的binding，将消息投递到一个或多个队列里。

注意：在声明一个队列后，如果将其持久化，则下次不需要进行声明。

3）Exchange说明

4、direct路由模式的使用

⽣产者发送消息时需要指定RoutingKey，即路由Key，Exchange接收到消息时转发到与RoutingKey相匹配的队列中。

1）生产者

import pika
import sys

credentials = pika.PlainCredentials('admin', 'admin') # mq⽤户名和密码
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',port=5672,virtual_host='/',credentials=credentials))
channel = connection.channel()
# 声明⼀个名为direct_logs的direct类型的exchange
channel.exchange_declare(exchange='direct_logs',
                         exchange_type='direct')

severity = ['info', 'warning', 'error']
for i in range(20):
    message = '{} Hello World! {}'.format(i, severity[i % 3])
    # 向名为direct_logs的exchage按照设置的routing_key发送messag
    channel.basic_publish(exchange='direct_logs',
                          routing_key=severity[i % 3],
                          body=message)
    print(" [x] Sent: {}".format(message))
connection.close()

2）消费者

import pika
import sys


credentials = pika.PlainCredentials('admin', 'admin') # mq⽤户名和密码
# BlockingConnection:同步模式
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost',port=5672,virtual_host='/',credentials=credentials))
channel = connection.channel()
# 在producer和consumer中分别声明⼀次以保证所要使⽤的exchange存在
channel.exchange_declare(exchange='direct_logs',exchange_type='direct')

result = channel.queue_declare(queue='cjavapy_d',durable=True)# ,arguments={"x-queue-type": "stream"})
# ⽤于获取临时queue的name
queue_name = result.method.queue
# Exchange就是根据这个RoutingKey和当前Exchange所有绑定的BindingKey做匹配，
channel.queue_bind(exchange='direct_logs',
                   queue=queue_name,
                   routing_key='info')#,arguments={"x-queue-type": "stream"})

print(' [*] Waiting for logs. To exit press CTRL+C')
# 定义⼀个回调函数来处理消息队列中的消息，
def callback(ch, method, properties, body):
# ⼿动发送确认消息
    print(" [x] %r:%r" % (method.routing_key, body))
    ch.basic_ack(delivery_tag=method.delivery_tag)
print(queue_name)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue=queue_name,on_message_callback=callback)
# 开始接收信息，并进⼊阻塞状态，队列⾥有信息才会调⽤callback进⾏处理
channel.start_consuming()

相关文档：

Python RabbitMQ pika的安装及work消息模型的使用

Python RabbitMQ pika的安装及单生产单消费模型的使用

Python RabbitMQ pika的安装及fanout消息订阅模式的使用