消息队列（Message Queue）是一种在分布式系统中进行异步通信的机制。它允许程序之间通过消息进行通信，而不需要直接调用对方。消息队列可以提高系统的可扩展性和容错能力，尤其是在处理高并发请求和异步任务时。
Python 中有几个常用的库可以用来实现消息队列，包括但不限于：
- RabbitMQ：使用 AMQP 协议的消息中间件。
- Redis：使用 Redis 的发布/订阅功能或列表数据结构实现消息队列。
- Kafka：高性能的分布式消息队列，常用于大数据处理。
- Celery：一个基于分布式消息传递的异步任务队列。
示例：使用 RabbitMQ
安装 RabbitMQ
首先，你需要安装 RabbitMQ 服务器。你可以从官方网站下载并安装 RabbitMQ，也可以使用包管理器（如 apt-get 或 yum）来安装。
安装 Pika 库
Pika 是一个 Python 库，用于与 RabbitMQ 交互。
发送消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| import pika
def send_message(message):
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='',
routing_key='hello',
body=message)
print(" [x] Sent %r" % message)
connection.close()
if __name__ == '__main__':
send_message('Hello World!')
|
接收消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| import pika
def receive_message():
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()
if __name__ == '__main__':
receive_message()
|
示例：使用 Redis
安装 Redis
确保你已经安装了 Redis 服务器。
安装 Redis 库
发送消息
1
2
3
4
5
6
7
8
9
| import redis
def send_message(message):
r = redis.Redis(host='localhost', port=6379, db=0)
r.lpush('queue:messages', message)
print(" [x] Sent %r" % message)
if __name__ == '__main__':
send_message('Hello World!')
|
接收消息
1
2
3
4
5
6
7
8
9
10
11
| import redis
def receive_message():
r = redis.Redis(host='localhost', port=6379, db=0)
while True:
message = r.rpop('queue:messages')
if message:
print(" [x] Received %r" % message.decode())
if __name__ == '__main__':
receive_message()
|
示例：使用 Celery
安装 Celery
配置 Celery
创建一个配置文件 celeryconfig.py：
1
2
| broker_url = 'redis://localhost:6379/0'
result_backend = 'redis://localhost:6379/0'
|
定义任务
创建一个任务文件 tasks.py：
1
2
3
4
5
6
7
| from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def add(x, y):
return x + y
|
发送任务
1
2
3
4
5
| from tasks import add
result = add.delay(4, 4)
print(result.ready())
print(result.get(timeout=1))
|
启动 Celery 工作者
在命令行中启动 Celery 工作者：
1
| celery -A tasks worker --loglevel=info
|
示例：使用 Kafka
安装 Kafka
确保你已经安装了 Kafka 服务器。
安装 Kafka-Python 库
1
| pip install kafka-python
|
发送消息
1
2
3
4
5
6
7
8
9
10
11
12
| from kafka import KafkaProducer
import json
def send_message(message):
producer = KafkaProducer(bootstrap_servers='localhost:9092',
value_serializer=lambda v: json.dumps(v).encode('utf-8'))
future = producer.send('my-topic', value=message)
result = future.get(timeout=60)
print(" [x] Sent %r" % message)
if __name__ == '__main__':
send_message({'key': 'value'})
|
接收消息
1
2
3
4
5
6
7
8
9
10
11
12
| from kafka import KafkaConsumer
import json
def receive_message():
consumer = KafkaConsumer('my-topic',
bootstrap_servers='localhost:9092',
value_deserializer=lambda m: json.loads(m.decode('utf-8')))
for message in consumer:
print(" [x] Received %r" % message.value)
if __name__ == '__main__':
receive_message()
|
总结
以上是几个常用的消息队列实现示例。选择哪种消息队列取决于你的具体需求和场景：
- RabbitMQ：适用于需要可靠的消息传递和复杂的路由规则的场景。
- Redis：适用于简单的消息队列，易于部署和使用。
- Kafka：适用于大数据处理和流式处理的场景。
- Celery：适用于异步任务队列和定时任务。
如果你有其他具体的需求或问题，请告诉我，我可以进一步帮助你。