Python操作RabbitMQ实战：安装、消息收发与高可用配置

沈亦白

发布时间：2026-10-13 | 3人浏览过

原创

本文详解如何在Linux环境下安装RabbitMQ，并利用Python的Pika库实现消息队列的基本操作。内容涵盖生产者与消费者的代码实现、消息确认机制（ACK）以防止数据丢失，以及通过持久化配置保障RabbitMQ宕机后的数据安全性。此外，还介绍了Fanout发布订阅模式的应用，帮助开发者构建高可用、解耦的分布式系统架构。

前言

RabbitMQ是基于AMQP协议构建的完整、可复用的企业级消息系统，遵循Mozilla Public License开源协议。消息队列（MQ）是一种应用间通信机制，应用程序通过读写队列中的消息进行数据交互，无需建立专用连接。这种消息传递方式不同于直接调用（如远程过程调用），它通过队列解耦了发送和接收过程，使得双方无需同时在线执行。

应用场景：

RabbitMQ是目前最主流的消息队列之一，支持多种编程语言，对于.NET开发者而言也值得了解。其典型应用场景主要包括三种：

1、系统集成与分布式架构设计。各子系统通过消息进行交互，这种模式逐渐演变为“基于消息传递的架构”风格。

2、提升系统吞吐量。当同步处理严重影响性能时（如日志记录），若同步写入日志会拖慢响应速度，可将日志消息发送至队列，由专门的子系统异步消费，从而保证主系统的响应速度。

3、保障系统高可用性。以电商秒杀场景为例，瞬时大量请求可能导致服务器或数据库宕机。将请求转发至消息队列，由服务器按能力逐步消费，可使流量削峰填谷，提高系统可用性。

一、安装环境

首先在Linux系统上安装RabbitMQ。

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

随后使用pip安装Python3的开发包。

pip3 install pika

软件安装完成后，访问http://115.xx.xx.xx:15672/即可进入RabbitMQ自带的Web管理界面进行查看和管理。默认管理员用户名和密码均为guest。

二、简单的向队列中加入消息

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

三、简单的从队列中获取消息

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

四、万一消费者掉线了

设想一种情况：

消费者从队列获取了n条数据，正在处理时发生宕机，该如何应对？RabbitMQ提供了ACK（确认机制）来确保消费者处理完毕。类似于网络中的ACK，消费者获取数据后，队列不会立即移除数据，而是等待对应的ACK。消费者处理完成后发送ACK包通知RabbitMQ，此时数据才会从队列中删除。因此，即使消费者掉线，数据仍保留在队列中，可供其他消费者处理。

在Python中实现如下：

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

五、万一 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,)) 来声明队列，才能确保数据持久化。

六、最简单的发布订阅

RabbitMQ中的最简单发布订阅称为Fanout模式。订阅者订阅特定频道，发布者向该频道发送消息，所有订阅者均可接收。由于发布者需使用订阅者创建的随机队列，因此必须先启动订阅者，再启动发布者。

发布者代码：

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

订阅者代码：

#!/usr/bin/env python3
# coding=utf-8
# @Time : 2017/6/13 20:20
# @Author : Shawn
# @Blog : https://blog.just666.cn
# @Email : shawnbluce@gmail.com
# @purpose : RabbitMQ_Subscriber
import pika
credentials = pika.PlainCredentials('guest', 'guest')
# 连接到RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('115.xx.xx.xx', 5672, '/', credentials))
channel = connection.channel()
# 定义交换机，进行exchange声明，exchange表示交换机名称，type表示类型
channel.exchange_declare(exchange='my_fanout',
       type='fanout')
# 随机创建队列
result = channel.queue_declare(exclusive=True) # exclusive=True表示建立临时队列，当consumer关闭后，该队列就会被删除
queue_name = result.method.queue
# 将队列与exchange进行绑定
channel.queue_bind(exchange='my_fanout',
     queue=queue_name)
# 定义回调方法
def callback(ch, method, properties, body):
 print(body.decode('utf-8'))
# 从队列获取信息
channel.basic_consume(callback,
      queue=queue_name,
      no_ack=True)
channel.start_consuming()

总结

以上就是本文的全部内容，希望这些内容能对您的学习或工作有所帮助。如有疑问，欢迎留言交流，感谢对本站的支持。