Python操作RabbitMQ：消息队列实战

02-08 1829阅读

在现代软件开发中，消息队列扮演着至关重要的角色。它能够有效地解耦系统组件，提高系统的可靠性和可扩展性。RabbitMQ是一款广泛使用的开源消息代理软件，而Python作为一种简洁高效的编程语言，与RabbitMQ结合可以实现强大的消息队列功能。

安装与配置

首先，确保你已经安装了RabbitMQ和Python的pika库。可以使用pip命令进行安装：

pip install pika

安装完成后，需要配置RabbitMQ。确保RabbitMQ服务已经启动并运行。

Python操作RabbitMQ：消息队列实战

发送消息

下面是一个简单的Python代码示例，用于向RabbitMQ发送消息：

import pika

# 创建连接
connection = pika.BlockingConnection(pika.ConnectionParameters('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()

这段代码首先创建了一个到本地RabbitMQ服务器的连接，然后声明了一个名为hello的队列。接着，使用basic_publish方法向队列发送了一条消息。

接收消息

接下来，编写接收消息的代码：

import pika

# 创建连接
connection = pika.BlockingConnection(pika.ConnectionParameters('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()

此代码创建连接并声明队列后，定义了一个回调函数callback来处理接收到的消息。然后使用basic_consume方法开始消费队列中的消息。

实战场景

假设我们有一个电商系统，订单服务和库存服务是两个独立的模块。当用户下单时，订单服务需要通知库存服务减少相应的库存。我们可以使用RabbitMQ来实现这个异步通信。

订单服务发送订单消息：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='order_queue')

order_info = {
    'product_id': 1,
    'quantity': 5
}
channel.basic_publish(exchange='', routing_key='order_queue', body=str(order_info))
print(" [x] Sent order information")

connection.close()

库存服务接收订单消息并处理库存减少：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='order_queue')

def process_order(ch, method, properties, body):
    order = eval(body)
    product_id = order['product_id']
    quantity = order['quantity']
    # 这里进行实际的库存减少逻辑，例如更新数据库
    print(f" [x] Processing order for product {product_id}, reducing stock by {quantity}")

channel.basic_consume(queue='order_queue', on_message_callback=process_order, auto_ack=True)

print(' [*] Waiting for order messages. To exit press CTRL+C')
channel.start_consuming()