Python脚本如何操作消息队列

本文目录导读：

1. Redis (简单队列)
2. RabbitMQ (专业消息队列)
3. Celery (分布式任务队列)
4. Kafka (高吞吐量消息队列)
5. ZeroMQ (轻量级消息队列)
6. 选择建议
7. 实际使用示例（使用RabbitMQ）

我来详细介绍Python操作消息队列的几种常见方式：

Redis (简单队列)

安装

pip install redis

生产者

import redis
import json
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 发送消息
def send_message(queue_name, message):
    r.lpush(queue_name, json.dumps(message))
# 使用示例
send_message('my_queue', {'task': 'process_data', 'data': [1, 2, 3]})

消费者

import redis
import json
import time
r = redis.Redis(host='localhost', port=6379, db=0)
def consume_message(queue_name):
    while True:
        # 阻塞等待消息
        message = r.brpop(queue_name, timeout=5)
        if message:
            _, data = message
            task = json.loads(data)
            print(f"处理任务: {task}")
            # 处理任务...
        else:
            print("没有新消息，等待中...")
        time.sleep(1)
consume_message('my_queue')

RabbitMQ (专业消息队列)

安装

pip install pika

生产者

import pika
# 连接RabbitMQ
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("消息已发送")
connection.close()

消费者

import pika
def callback(ch, method, properties, body):
    print(f"收到消息: {body}")
# 连接
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()
# 声明队列
channel.queue_declare(queue='hello')
# 消费消息
channel.basic_consume(
    queue='hello',
    on_message_callback=callback,
    auto_ack=True
)
print('等待消息...')
channel.start_consuming()

Celery (分布式任务队列)

安装

pip install celery redis

配置文件 celery_app.py

from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def add(x, y):
    return x + y
@app.task
def process_data(data):
    print(f"处理数据: {data}")
    return len(data)

启动Worker

celery -A celery_app worker --loglevel=info

发送任务

from celery_app import add, process_data
# 同步调用
result = add.delay(4, 5)
print(f"结果: {result.get()}")
# 异步调用
result = process_data.delay([1, 2, 3, 4, 5])
print(f"任务ID: {result.id}")

Kafka (高吞吐量消息队列)

安装

pip install kafka-python

生产者

from kafka import KafkaProducer
import json
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
# 发送消息
producer.send('my_topic', {'key': 'value', 'number': 42})
producer.flush()
print("消息已发送")

消费者

from kafka import KafkaConsumer
import json
consumer = KafkaConsumer(
    'my_topic',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)
for message in consumer:
    print(f"收到消息: {message.value}")

ZeroMQ (轻量级消息队列)

安装

pip install pyzmq

生产者

import zmq
context = zmq.Context()
socket = context.socket(zmq.PUSH)
socket.bind("tcp://*:5555")
# 发送消息
socket.send_json({"task": "compute", "data": [1, 2, 3]})
print("消息已发送")

消费者

import zmq
context = zmq.Context()
socket = context.socket(zmq.PULL)
socket.connect("tcp://localhost:5555")
while True:
    message = socket.recv_json()
    print(f"收到消息: {message}")

选择建议

实际使用示例（使用RabbitMQ）

import pika
import json
from datetime import datetime
class MessageQueue:
    def __init__(self, host='localhost'):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host)
        )
        self.channel = self.connection.channel()
    def declare_queue(self, queue_name):
        self.channel.queue_declare(queue=queue_name, durable=True)
    def publish(self, queue_name, message):
        self.channel.basic_publish(
            exchange='',
            routing_key=queue_name,
            body=json.dumps(message),
            properties=pika.BasicProperties(
                delivery_mode=2,  # 持久化消息
            )
        )
    def consume(self, queue_name, callback):
        self.channel.basic_consume(
            queue=queue_name,
            on_message_callback=callback,
            auto_ack=True
        )
        self.channel.start_consuming()
    def close(self):
        self.connection.close()
# 使用示例
def process_message(ch, method, properties, body):
    data = json.loads(body)
    print(f"[{datetime.now()}] 处理消息: {data}")
mq = MessageQueue()
mq.declare_queue('task_queue')
# 发送消息
mq.publish('task_queue', {'id': 1, 'action': 'process'})
# 消费消息（在另一个脚本中）
mq.consume('task_queue', process_message)

选择的依据：

• 简单需求：使用Redis或ZeroMQ
• 企业级应用：使用RabbitMQ
• 大数据处理：使用Kafka
• 异步任务：使用Celery