redis怎么实现延迟队列 redis延迟队列的2种实现方式

import redis
import time
import json

class DelayQueue:
    def __init__(self, redis_host='localhost', redis_port=6379, queue_name='delay_queue'):
        self.redis_client = redis.Redis(host=redis_host, port=redis_port)
        self.queue_name = queue_name

    def enqueue(self, task, delay_seconds):
        """
        将任务加入延迟队列
        :param task: 任务内容 (可以是字典或字符串)
        :param delay_seconds: 延迟时间 (秒)
        """
        execute_time = time.time() + delay_seconds
        task_json = json.dumps(task)  # 将任务序列化为 JSON 字符串
        self.redis_client.zadd(self.queue_name, {task_json: execute_time})


    def dequeue(self):
        """
        从延迟队列中取出到期任务
        :return: 到期任务 (如果队列为空，则返回 None)
        """
        now = time.time()
        tasks = self.redis_client.zrangebyscore(self.queue_name, min=0, max=now, start=0, num=1) # 获取一个
        if tasks:
            task_json = tasks[0]
            try:
                task = json.loads(task_json) # 反序列化
                self.redis_client.zrem(self.queue_name, task_json) # 移除任务
                return task
            except json.JSONDecodeError:
                print(f"Error decoding JSON: {task_json}")
                return None
        else:
            return None

# 使用示例
if __name__ == '__main__':
    delay_queue = DelayQueue()
    task1 = {"message": "Task 1", "priority": "high"}
    task2 = {"message": "Task 2", "priority": "low"}

    delay_queue.enqueue(task1, 5)  # 5秒后执行
    delay_queue.enqueue(task2, 10) # 10秒后执行

    print("Tasks enqueued.  Waiting for execution...")

    while True:
        task = delay_queue.dequeue()
        if task:
            print(f"Executing task: {task}")
        else:
            time.sleep(1) # 避免空转，降低 CPU 占用

import redis
import time
import threading

class DelayedTaskHandler:
    def __init__(self, redis_host='localhost', redis_port=6379, queue_name='delayed_tasks'):
        self.redis_client = redis.Redis(host=redis_host, port=redis_port)
        self.pubsub = self.redis_client.pubsub()
        self.queue_name = queue_name
        self.task_channel = '__keyevent@0__:expired' # 监听过期事件

    def enqueue(self, task_id, task_data, delay_seconds):
        """将任务添加到 Redis 列表，并设置过期时间"""
        task = {'id': task_id, 'data': task_data}
        key = f"{self.queue_name}:{task_id}"  # 构造唯一的 key
        self.redis_client.setex(key, delay_seconds, str(task))  # 使用 SETEX 设置键和过期时间


    def process_expired_task(self, message):
        """处理过期的任务"""
        key = message['data'].decode('utf-8')
        if key.startswith(self.queue_name): # 确保是我们的任务
            try:
                task_data = self.redis_client.get(key) # 理论上已经过期，但为了保险
                if task_data:
                    print(f"Processing expired task: {task_data.decode('utf-8')}")
                    self.redis_client.delete(key)  # 删除键（如果还存在）
            except Exception as e:
                print(f"Error processing task: {e}")

    def run(self):
        """监听过期事件并处理任务"""
        self.pubsub.psubscribe(self.task_channel)
        for message in self.pubsub.listen():
            if message['type'] == 'pmessage':
                self.process_expired_task(message)

if __name__ == '__main__':
    handler = DelayedTaskHandler()

    # 创建一个线程来监听过期事件
    listener_thread = threading.Thread(target=handler.run)
    listener_thread.daemon = True  # 设置为守护线程
    listener_thread.start()

    # 模拟添加一些延迟任务
    handler.enqueue("task1", {"payload": "Do something important!"}, 5) # 5秒后执行
    handler.enqueue("task2", {"payload": "Another important task"}, 10) # 10秒后执行


    print("Tasks enqueued.  Waiting for tasks to expire...")
    time.sleep(15) # 等待一段时间让任务过期