redis延时队列zset实现的示例

import redis  
 import time  
   
 # 连接到Redis  
 r = redis.Redis(host='localhost', port=6379, db=0)  
   
 # 假设有一个任务需要在30秒后执行  
 task_id = 'task_123'  
 execute_at = int(time.time()) + 30  # 当前时间加上延时秒数  
   
 # 将任务添加到ZSet中  
 r.zadd('delay_queue', {task_id: execute_at})

def process_delay_queue(r, queue_key):  
     # 获取当前时间戳  
     now = int(time.time())  
       
     # 使用ZRANGEBYSCORE和ZREM命令结合Lua脚本来原子性地检查并移除到期的任务  
     # 这里为了简化，直接用Python代码模拟，实际生产环境建议使用Lua脚本以保证原子性  
     while True:  
         # 查找并移除分数小于等于当前时间戳的任务  
         # 注意：这里ZRANGEBYSCORE返回的是成员列表，需要再调用ZREM移除  
         # 但由于ZRANGEBYSCORE+ZREM不是原子操作，实际使用时推荐使用Lua脚本  
         ready_tasks = r.zrangebyscore(queue_key, 0, now, withscores=True)  
         if not ready_tasks:  
             break  # 没有到期的任务，跳出循环  
   
         for task_id, _ in ready_tasks:  
             # 假设这里执行任务  
             print(f"Executing task: {task_id}")  
               
             # 移除已执行的任务  
             r.zrem(queue_key, task_id)  
   
 # 调用函数处理延时队列  
 process_delay_queue(r, 'delay_queue')

-- Lua脚本，用于原子性地查找并移除到期的任务  
 local queue_key = KEYS[1]  
 local now = ARGV[1]  
   
 -- 使用redis.call执行Redis命令  
 local ready_tasks = redis.call('ZRANGEBYSCORE', queue_key, 0, now)  
 if #ready_tasks > 0 then  
     redis.call('ZREM', queue_key, unpack(ready_tasks))  
     -- 这里可以添加逻辑来实际执行任务，但Lua脚本中通常不推荐进行复杂的逻辑处理  
     -- 可以将任务ID返回给客户端，由客户端执行具体的任务逻辑  
     return ready_tasks  
 else  
     return nil  
 end

# 加载并执行Lua脚本（略去具体加载脚本的代码）  
 # ...  
   
 # 调用Lua脚本处理延时队列  
 now = int(time.time())  
 script_result = r.eval（script, 1, 'delay_queue', now)  
 if script_result:  
     # 处理返回的到期任务列表  
     for task_id in script_result:  
         print(f"Executing task (Lua script): {task_id}")