使用 Python 操作消息队列 RabbitMQ

0X00 安装环境

首先是在Linux上安装rabbitmq

1# 环境为CentOS 7
2yum install rabbitmq-server	# 安装RabbitMQ
3systemctl start rabbitmq-server	# 启动
4systemctl enable rabbitmq-server	# 开机自启
5systemctl stop firewall-cmd		# 临时关闭防火墙

然后用pip安装Python3的开发包

1pip3 install pika

安装好软件之后可以访问http://115.xx.xx.xx:15672/来访问自带的web页面来查看和管理RabbitMQ。默认管理员的用户密码都是guest

0X01 简单的向队列中加入消息

python

1#!/usr/bin/env python3
2# coding=utf-8
3# @Time    : 2017/6/13 19:25
4# @Author  : Shawn
5# @Blog    : https://blog.just666.cn
6# @Email   : shawnbluce@gmail.com
7# @purpose : RabbitMQ_Producer
8
9import pika
10
11# 创建连接对象
12connection = pika.BlockingConnection(pika.ConnectionParameters(host='115.xx.xx.xx'))
13
14# 创建频道对象
15channel = connection.channel()
16
17# 指定一个队列，如果该队列不存在则创建
18channel.queue_declare(queue='test_queue')
19
20# 提交消息
21for i in range(10):
22    channel.basic_publish(exchange='', routing_key='test_queue', body='hello,world' + str(i))
23    print("sent...")
24
25# 关闭连接
26connection.close()

0X02 简单的从队列中获取消息

python

1#!/usr/bin/env python3
2# coding=utf-8
3# @Time    : 2017/6/13 19:40
4# @Author  : Shawn
5# @Blog    : https://blog.just666.cn
6# @Email   : shawnbluce@gmail.com
7# @purpose : RabbitMQ_Consumer
8
9import pika
10
11credentials = pika.PlainCredentials('guest', 'guest')
12
13# 连接到RabbitMQ服务器
14connection = pika.BlockingConnection(pika.ConnectionParameters('115.xx.xx.xx', 5672, '/', credentials))
15channel = connection.channel()
16
17# 指定一个队列，如果该队列不存在则创建
18channel.queue_declare(queue='test_queue')
19
20# 定义一个回调函数
21def callback(ch, method, properties, body):
22    print(body.decode('utf-8'))
23
24
25# 告诉RabbitMQ使用callback来接收信息
26channel.basic_consume(callback, queue='test_queue', no_ack=False)
27
28print('waiting...')
29
30# 开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理。按ctrl+c退出。
31channel.start_consuming()

0X03 万一消费者掉线了

想象这样一种情况：

消费者从消息队列中获取了n条数据，正要处理呢结果宕机了，那该怎么办？在RabbieMQ中有一个ACK可以用来确认消费者处理结束。就有点类似网络中的ACK，消费者每次从队列中获取了数据之后队列不会立刻将数据移除，而是等待对应的ACK。消费者获取到数据并处理完成之后会向队列发送一个ACK包，通知RabbitMQ这堆消息已经处理妥当了，可以删除了，这时候RabbitMQ才会将数据从队列中移除。所以这种情况下即使消费者掉线也没有什么问题，数据依旧会在队列中存在，留给其他消费者处理。

在Python中这样实现：

消费者有这样一行代码channel.basic_consume(callback, queue='test_queue', no_ack=False)，其中no_ack=False表示不发送确认包。将其修改为no_ack=True就会在每次处理完之后向RabbitMQ发送一个确认包，以确认消息处理完毕。

0X04 万一RabbitMQ宕机了呢

虽然有了ACK包，但是万一RabbitMQ挂了那数据还是会损失。所以我们可以给RabbitMQ设置一个数据持久化存储。RabbitMQ会将数据持久化存储到磁盘上，保证下次再启动的时候队列还在。

在Python中这样实现：

我们声明一个队列是这样的channel.queue_declare(queue='test_queue')，如果需要持久化一个队列可以这样声明channel.queue_declare(queue='test_queue', durable=True)。不过这行直接放在代码中是不能执行的，因为以前已经有了一个名为test_queue的队列，RabbitMQ不允许用不同的方式声明同一个队列，所以可以换一个队列名新建来指定数据持久化存储。不过如果只是这样声明的话，在RabbitMQ宕机重启后确实队列还在，不过队列里的数据就没有了。除非我们这样来声明队列channel.basic_publish(exchange='', routing_key="test_queue", body=message, properties=pika.BasicProperties(delivery_mode = 2,))。

0X05 最简单的发布订阅

最简单的发布订阅在RabbitMQ中称之为Fanout模式。也就是说订阅者订阅某个频道，然后发布者向这个频道中发布消息，所有订阅者就都能接收到这条消息。不过因为发布者需要使用订阅者创建的随机队列所以需要先启动订阅者才能启动发布者。

发布者代码：

python

1#!/usr/bin/env python3
2# coding=utf-8
3# @Time    : 2017/6/13 20:21
4# @Author  : Shawn
5# @Blog    : https://blog.just666.cn
6# @Email   : shawnbluce@gmail.com
7# @purpose : RabbitMQ_Publisher
8
9import pika
10
11# 创建连接对象
12connection = pika.BlockingConnection(pika.ConnectionParameters(host='115.xx.xx.xx'))
13
14# 创建频道对象
15channel = connection.channel()
16
17# 定义交换机，exchange表示交换机名称，type表示类型
18channel.exchange_declare(exchange='my_fanout',
19                         type='fanout')
20
21message = 'Hello Python'
22# 将消息发送到交换机
23channel.basic_publish(exchange='my_fanout',  # 指定exchange
24                      routing_key='',  # fanout下不需要配置，配置了也不会生效
25                      body=message)
26connection.close()

订阅者代码：

python

1#!/usr/bin/env python3
2# coding=utf-8
3# @Time    : 2017/6/13 20:20
4# @Author  : Shawn
5# @Blog    : https://blog.just666.cn
6# @Email   : shawnbluce@gmail.com
7# @purpose : RabbitMQ_Subscriber
8
9import pika
10
11credentials = pika.PlainCredentials('guest', 'guest')
12
13# 连接到RabbitMQ
14connection = pika.BlockingConnection(pika.ConnectionParameters('115.xx.xx.xx', 5672, '/', credentials))
15channel = connection.channel()
16
17# 定义交换机，进行exchange声明，exchange表示交换机名称，type表示类型
18channel.exchange_declare(exchange='my_fanout',
19                         type='fanout')
20
21# 随机创建队列
22result = channel.queue_declare(exclusive=True)  # exclusive=True表示建立临时队列，当consumer关闭后，该队列就会被删除
23queue_name = result.method.queue
24
25# 将队列与exchange进行绑定
26channel.queue_bind(exchange='my_fanout',
27                   queue=queue_name)
28
29# 定义回调方法
30def callback(ch, method, properties, body):
31    print(body.decode('utf-8'))
32
33
34# 从队列获取信息
35channel.basic_consume(callback,
36                      queue=queue_name,
37                      no_ack=True)
38
39channel.start_consuming()