当前位置：首页 > article >正文

RabbitMQ 实现消息队列负载均衡

article 2024/11/7 9:36:51

在现代应用程序中，消息队列是一种重要的架构模式，用于解耦服务、处理异步任务和实现负载均衡。其中，RabbitMQ是一个广泛使用的开源消息代理，提供了高可用性、可靠性和灵活性。本文将展示如何使用Python及其pika库来实现RabbitMQ，并构建一个简单的生产者-消费者模型。

什么是RabbitMQ？

RabbitMQ是一个消息代理，支持多种消息协议，尤其以AMQP协议著称。它允许应用程序在异步模式下交换数据，适合于微服务架构中的消息传递、任务调度等场景。

安装依赖

在开始之前，确保你已经安装了RabbitMQ服务器，并且在你的Python环境中安装了pika库。可以使用以下命令安装pika：

pip install pika

创建生产者（Producer）

生产者是负责将消息发送到队列的应用程序。以下是一个简单的Python生产者示例：

import pika

# 创建连接到RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 声明一个持久化队列
channel.queue_declare(queue='task_queue', durable=True)

# 发送消息
for i in range(10):
    message = f'Message {i}'
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=message,
        properties=pika.BasicProperties(
            delivery_mode=2,  # 消息持久化
        ))
    print(f" [x] Sent '{message}'")

# 关闭连接
connection.close()

代码解释：

pika.BlockingConnection：建立与RabbitMQ的连接。
queue_declare：声明一个队列，如果队列不存在则创建它。
basic_publish：向队列发布消息。

创建消费者（Consumer）

消费者是从队列中读取消息并处理它们的应用程序。以下是一个简单的Python消费者示例：

import pika
import time

def callback(ch, method, properties, body):
    print(f" [x] Received '{body.decode()}'")
    time.sleep(body.count(b'.'))  # 模拟处理时间
    print(" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 创建连接到RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 声明同样的队列
channel.queue_declare(queue='task_queue', durable=True)

# 设置为公认模式，确保负载均衡
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)

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