Redis 与 RabbitMQ 实现延时队列的示例代码

在现代应用程序开发中，延时队列是一种常见的需求。它允许我们在指定的时间间隔后处理任务或消息。Redis 和 RabbitMQ 是两个强大的工具，结合它们可以有效地实现延时队列。以下是使用 Redis 和 RabbitMQ 实现延时队列的示例代码。

我们来看看 Redis 在延时队列中的作用。Redis 可以用于存储延时任务的相关信息，比如任务的标识、执行时间等。

import redis
import time

redis_client = redis.Redis()

def add_task_to_redis(task_id, delay_seconds):
    execute_time = time.time() + delay_seconds
    redis_client.zadd('delayed_tasks', {task_id: execute_time})

接下来，使用 RabbitMQ 来处理实际的任务分发。

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='task_queue')

def process_task(task_id):
    # 实际的任务处理逻辑
    print(f"Processing task {task_id}")

def consume_tasks():
    def callback(ch, method, properties, body):
        task_id = body.decode()
        process_task(task_id)

    channel.basic_consume(queue='task_queue', on_message_callback=callback, auto_ack=True)
    channel.start_consuming()

然后，我们需要一个定时的任务来从 Redis 中取出即将执行的任务，并将其放入 RabbitMQ 队列中。

import threading

def check_delayed_tasks():
    while True:
        current_time = time.time()
        tasks = redis_client.zrangebyscore('delayed_tasks', 0, current_time)
        for task_id in tasks:
            redis_client.zrem('delayed_tasks', task_id)
            channel.basic_publish(exchange='', routing_key='task_queue', body=task_id)
        time.sleep(1)

thread = threading.Thread(target=check_delayed_tasks)
thread.start()

通过以上代码，我们实现了一个简单的基于 Redis 和 RabbitMQ 的延时队列。Redis 负责存储和管理延时任务的时间信息，而 RabbitMQ 则负责任务的分发和处理。

在实际应用中，还需要根据具体的需求进行更多的错误处理、优化和扩展。例如，可以考虑使用 Redis 的事务来保证操作的原子性，或者增加任务的优先级等功能。

Redis 和 RabbitMQ 的结合为实现延时队列提供了一种高效、可靠的解决方案，能够满足各种复杂业务场景的需求。