Redis实现队列的阻塞、延时、发布和订阅

1. 阻塞队列

Redis提供了BRPOP命令实现阻塞队列。该命令用于在列表中阻塞并等待接收列表中的项。当列表为空时，客户端将一直等待，直到接收到一条数据或超时返回。BRPOP命令的语法如下：

BRPOP key [key ...] timeout

其中，key参数表示需要阻塞的队列名称，可以指定多个，以逐个查找队列；timeout参数是阻塞的时长，单位为秒，当时间耗尽时，BRPOP会返回NULL。示例代码如下：

rds = redis.Redis(host='localhost', port=6379)
res = rds.brpop('myqueue', timeout=10)
if res is not None:
    print(res[1])
else:
    print('timeout')

2. 延时队列

Redis中常用的延时队列实现方式是将消息放入zset有序集合中，设置其score为消息需要执行的时间。通过定期轮询有序集合，获取需要执行的任务进行处理。示例代码如下：

import time

rds = redis.Redis(host='localhost', port=6379)
while True:
    cur_time = time.time()
    res = rds.zrangebyscore('tasks', 0, cur_time, start=0, num=1)
    if len(res) > 0:
        task = res[0]
        rds.zrem('tasks', task)  # 从延时队列中移除已完成的任务
        print(f'Processing task: {task}')
    time.sleep(1)

3. 发布/订阅

Redis提供了发布/订阅功能，允许多个客户端订阅同一个频道（channel），当该频道有消息发布时，所有订阅者都会收到消息。Redis发布消息使用PUBLISH命令，订阅者使用SUBSCRIBE命令。示例代码如下：

# 发布者
import time

rds = redis.Redis(host='localhost', port=6379)
for i in range(5):
    time.sleep(1)
    rds.publish('mychannel', f'message_{i+1}')

# 订阅者
import redis

rds = redis.Redis(host='localhost', port=6379)
p = rds.pubsub()
p.subscribe('mychannel')
for message in p.listen():
    if message['type'] == 'message':
        print(message['data'])

在订阅时，可以通过参数指定订阅多个频道，也可以使用通配符进行订阅，例如：

p.subscribe('chan1', 'chan2', 'chan3.*')