怎么用redis+python做消息队列

使用Redis+Python实现消息队列

Redis是一个高性能的键值对存储系统，可以作为消息队列的实现工具之一。其提供了list、set、zset等多种数据结构，其中，list结构可以用来实现简单的队列，set和zset结构可以用来实现高级队列。配合Python脚本，可以实现消息的发送、接收、处理、存储等多个功能。

本文将介绍如何使用Redis和Python建立一个简单的消息队列。

1. 安装配置Redis

首先需要在本地安装Redis，根据操作系统的不同，安装方法会有所不同。以ubuntu为例，可以使用以下命令进行安装：

sudo apt-get install redis-server

此外，还需要安装redis的Python客户端redis-py，可以使用pip进行安装：

pip install redis

2. 使用Redis作为消息队列

在Redis中，可以使用list结构模拟队列。通过LPUSH命令将消息压入队列，通过BRPOP命令从队列中取出消息。以下是Python中使用redis-py库操作队列的示例代码：

import redis

# 建立Redis连接
r = redis.Redis(host='localhost', port=6379, db=0)
# 将消息加入队列
r.lpush('myqueue', 'hello world!')
# 从队列中取出消息
msg = r.brpop('myqueue')
print(msg[1])

以上代码的执行过程如下：

1.建立Redis连接；

2.使用lpush命令向名为'myqueue'的队列中加入一条消息'hello world!'；

3.使用brpop命令从'myqueue'队列中取出一条消息；

4.打印消息内容。

2.1 支持多个消费者同时获取消息

通过上面的代码，可以实现一个生产者和一个消费者之间的消息传递。但是，在实际应用中，可能会存在多个消费者同时获取消息的情况。这时，使用BRPOP命令会存在问题：BRPOP在取到队列中的一条消息之后，会将整个队列锁定，其他消费者无法读取队列中的其他消息。

因此，应该使用BLPOP命令来替代BRPOP命令。BLPOP命令可以实现多个消费者同时获取到队列中的消息，而不会出现其他消费者无法读取消息的情况。

以下是使用redis-py库实现多个消费者同时获取队列消息的示例代码：

import redis

import threading
def consumer(name: str):
    """消费者线程"""
    r = redis.Redis(host='localhost', port=6379, db=0)
    while True:
        msg = r.blpop('myqueue')
        print(f'{name} 消费了一条消息：{msg[1]}')def main():
    """生产者线程"""
    r = redis.Redis(host='localhost', port=6379, db=0)
    r.lpush('myqueue', 'hello world1')
    r.lpush('myqueue', 'hello world2')
    r.lpush('myqueue', 'hello world3')
    # 创建消费者线程
    threads = [threading.Thread(target=consumer, args=(f'consumer{i}',)) for i in range(5)]
    # 启动线程
    [t.start() for t in threads]
    # 等待所有线程执行完毕
    [t.join() for t in threads]if __name__ == "__main__":
    main()

以上代码的执行过程如下：

1.生产者线程向队列中添加三条消息；

2.创建五个消费者线程，并启动；

3.每个消费者线程不断从队列中获取消息，并处理。

2.2. 处理消息

在实际应用中，需要对从消息队列中获取到的消息进行处理。在处理消息的过程中，可能会有一些耗时的操作，比如数据验证、文件读写、网络请求等。如果直接在消费者线程中进行这些操作，会导致其他消息的处理会被阻塞，因此，应该使用多线程或多进程来处理消息。

以下是使用redis-py库和threading模块实现多线程消费消息的示例代码：

import redis

import time
import threading
def consumer(name: str):
    """消费者线程"""
    r = redis.Redis(host='localhost', port=6379, db=0)
    while True:
        msg = r.blpop('myqueue')
        data = msg[1].decode('utf-8')
        print(f'{name} 消费了一条消息：{data}')
        # 模拟耗时操作
        time.sleep(2)
        print(f'{name} 消费了消息：{data} 完成')def main():
    """生产者线程"""
    r = redis.Redis(host='localhost', port=6379, db=0)
    r.lpush('myqueue', 'hello world1')
    r.lpush('myqueue', 'hello world2')
    r.lpush('myqueue', 'hello world3')
    # 创建消费者线程
    threads = [threading.Thread(target=consumer, args=(f'consumer{i}',)) for i in range(3)]
    # 启动线程
    [t.start() for t in threads]
    # 等待所有线程执行完毕
    [t.join() for t in threads]
if __name__ == "__main__":
    main()

以上代码的执行过程如下：

1.生产者线程向队列中添加三条消息；

2.创建三个消费者线程，并启动；

3.每个消费者线程不断从队列中获取消息，并模拟处理过程。

3. 使用Redis实现延时队列

除了普通的消息队列之外，还可以使用Redis实现延时队列。在延时队列中，消息不会立即被发送，而是需要等待一定的时间之后再被发送。这种机制可以满足一些特定的业务需求: 如订单超时未支付等。

Redis提供了zset结构，可以方便地实现延时队列。在zset中，每个元素由一个score和一个value组成。score用于按照时间轴排序，value表示消息内容。通过使用zadd命令将消息添加到zset中，使用zrangebyscore命令获取待发送的消息，并使用zrem命令将已经发送的消息从zset中删除。

下面是使用Redis实现延时队列的示例代码：

import redis

import threading
import time
def consumer(name: str):
    """消费者线程"""
    r = redis.Redis(host='localhost', port=6379, db=0)
    while True:
        # 获取当前时间戳
        score = int(time.time())
        # 尝试获取score小于当前时间戳的消息
        msgs = r.zrangebyscore('delay_queue', 0, score)
        if not msgs:
            time.sleep(1)
            continue
        print(f'当前消息数：{len(msgs)}')
        # 删除已经到期的消息
        r.zrem('delay_queue', *msgs)
        # 处理消息
        for msg in msgs:
            print(f'{name} 消费了一条消息：{msg.decode("utf-8")}')
def producer():
    """生产者线程"""
    r = redis.Redis(host='localhost', port=6379, db=0)
    msg = 'Hello, Delayed World!'
    # 发送一条延时消息
    r.zadd('delay_queue', {msg: time.time() + 5})
    print(f'{msg} 已添加到延时队列')
if __name__ == "__main__":
    producer()
    # 创建消费者线程
    threads = [threading.Thread(target=consumer, args=(f'consumer{i}',)) for i in range(3)]
    # 启动线程
    [t.start() for t in threads]
    # 等待所有线程执行完毕
    [t.join() for t in threads]

以上代码的执行过程如下：

1.生产者线程向延时队列中添加一条消息；

2.创建三个消费者线程，并启动；

3.消费者线程轮询查找延时队列中已经到期的消息，并处理。

4. 总结

通过Redis和Python，我们可以轻松地建立一个简单的消息队列系统。Redis作为一个轻量级的高性能数据库，可以提供快速的队列操作，满足大部分中小型应用场景的需求。

当然，Redis也有一些限制，例如不能保证消息的顺序等。对于复杂的应用场景，可能需要使用更为专业的消息队列中间件，如rabbitmq等。

最后建议在实际开发中，结合具体的业务需求和系统环境，选择适合的消息队列方案。