基于RabbitMQ的异步消息传递：发送与消费

news2024/11/26 9:15:07

引言

RabbitMQ是一个流行的开源消息代理，用于在分布式系统中实现异步消息传递。它基于Erlang语言编写，具有高可用性和可伸缩性。在本文中，我们将探讨如何在Python中使用RabbitMQ进行消息发送和消费。

安装RabbitMQ

在 Ubuntu 上安装 RabbitMQ 可以通过多种方式完成，包括使用包管理器、Docker 容器或从源代码编译。以下是最简单和最常见的方法，使用包管理器进行安装。

安装 Erlang:
添加 PPA 之后，可以安装 Erlang。

sudo apt install erlang

安装 RabbitMQ:
在 Erlang 安装完成后，可以安装 RabbitMQ。

sudo apt install rabbitmq-server

启动 RabbitMQ:
安装完成后，可以启动 RabbitMQ 服务。

sudo systemctl start rabbitmq-server

设置 RabbitMQ 用户:
为了安全起见，应该创建一个 RabbitMQ 用户。

sudo rabbitmqctl add_user your_username your_password
sudo rabbitmqctl set_permissions -p / your_username ".*" ".*" ".*"
sudo rabbitmqctl set_user_tags your_username administrator
sudo rabbitmqctl set_permissions -p / your_username ".*" ".*" ".*"

在这里，your_username 是想要创建的用户名，your_password 是该用户的密码。

检查 RabbitMQ 状态:
使用以下命令检查 RabbitMQ 服务状态。

sudo systemctl status rabbitmq-server

配置 RabbitMQ 防火墙:
根据您的网络配置，可能需要配置防火墙规则以允许外部客户端访问 RabbitMQ。

sudo ufw allow from any to any port 5672 proto tcp
sudo ufw allow from any to any port 15672 proto tcp

停止 RabbitMQ:
如果想要停止 RabbitMQ 服务，可以使用以下命令。

sudo systemctl stop rabbitmq-server

查看 RabbitMQ 版本:
使用以下命令查看已安装的 RabbitMQ 版本。

sudo rabbitmq-server -v

安装完成后，可以通过访问 http://localhost:15672 来访问 RabbitMQ 管理界面。如果已经设置了用户，将需要使用创建的用户名和密码登录。请注意，RabbitMQ 服务器配置和安全性是复杂的主题，上述步骤提供了基本的安装和配置指南。根据具体需求，可能需要进行更详细的配置。

安装pika

pika 是一个用于 RabbitMQ 的 Python 客户端库，它允许创建和控制 RabbitMQ 队列、交换器、绑定和消息。安装也非常简单，安装完成后，就可以在 Python 代码中导入 pika 库并使用它来与 RabbitMQ 交互。

pip install pika

发送消息

首先，来看一下如何发送消息到RabbitMQ队列。以下代码片段展示了如何连接到RabbitMQ服务器，声明一个队列，并发布一个消息到该队列。

#!/usr/bin/env python
import pika


params = pika.ConnectionParameters(
    host='localhost', heartbeat=3600, blocked_connection_timeout=300)
connection = pika.BlockingConnection(
     pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()

创建一个ConnectionParameters实例，定义了与RabbitMQ服务器建立连接所需的参数：
- host='localhost'：指定RabbitMQ服务器的主机名，这里是本地主机。
- heartbeat=3600：心跳间隔，单位为秒，用于保持连接的活跃性。
- blocked_connection_timeout=300：如果连接被阻塞，这个参数定义了连接超时的时间，单位为秒。
connection = pika.BlockingConnection(...)：使用pika.BlockingConnection创建一个到RabbitMQ的阻塞连接。这意味着连接操作会等待直到成功建立连接。
channel = connection.channel()：创建一个新的通信信道。在RabbitMQ中，信道是进行消息传递的通道。
channel.queue_declare(queue='hello')：声明一个名为hello的队列。如果该队列不存在，RabbitMQ会创建它。
channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')：发布（发送）一条消息到队列。参数说明：
- exchange：交换机名称，这里为空字符串，表示使用默认的交换机。
- routing_key：路由键，这里与队列名相同，表示消息将直接发送到hello队列。
- body：消息体，这里是字符串'Hello World!'。

消费消息

接下来，看一下如何从RabbitMQ队列中消费消息。以下代码片段展示了如何连接到RabbitMQ服务器，声明一个队列，并使用回调函数来处理收到的消息。

#!/usr/bin/env python
import pika, sys, os


def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='hello')
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)

    channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

定义一个名为callback的函数，它将作为消费消息时的回调函数。当消息到达时，这个函数会被调用，并打印出消息体。
channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)：开始消费hello队列中的消息。on_message_callback参数指定了当消息到达时调用的回调函数，auto_ack=True表示自动确认消息。