4. 任务队列与消息中间件：Redis队列、RabbitMQ、Kafka在调度中的应用、延迟队列实现

聊到任务调度，队列是个绕不开的话题。说白了，调度系统就是个「生产者-消费者」模型——任务生产者往里塞任务，调度器从里面捞任务执行。那这个「中间容器」用什么？我见过不少团队直接拿数据库表当队列用，嗯，小规模还行，一旦量上来，锁竞争、延迟飙升，分分钟教你做人。

今天咱们就聊聊三种主流的队列方案：Redis、RabbitMQ、Kafka。它们各有脾气，选对了是利器，选错了……我踩过坑，后面会讲。

4.1 Redis队列：轻量级选手的极限

Redis做队列，最直观的就是用 List 结构。LPUSH 塞任务，RPOP 取任务，简单粗暴。

# 生产者
LPUSH task_queue "{'id': 1001, 'type': 'send_email'}"

# 消费者
RPOP task_queue

但有个问题——如果队列空了，消费者就得轮询。轮询太频繁，CPU 白烧；太慢，任务延迟高。Redis 提供了 BRPOP 阻塞读取，解决了这个问题：

# 阻塞式消费，超时30秒
BRPOP task_queue 30

我在项目中遇到过一个小团队，用 Redis List 做延迟任务队列。他们搞了个死循环，每秒轮询一次，结果 Redis CPU 冲到 90%。后来改成 BRPOP，CPU 直接降到 5%。

4.2 RabbitMQ：可靠性的守护者

如果你需要「消息必达」，RabbitMQ 是更稳妥的选择。它支持消息持久化、ACK 确认、死信队列，调度系统里最常见的用法是：

• Direct Exchange：精确路由，适合指定任务类型
• Topic Exchange：通配符匹配，适合多维度任务分发
• Fanout Exchange：广播，适合所有节点都执行的任务

举个例子，一个任务调度系统里，不同优先级的任务走不同队列：

# 声明队列
channel.queue_declare(queue='high_priority', durable=True)
channel.queue_declare(queue='normal_priority', durable=True)

# 生产者发送
channel.basic_publish(
    exchange='',
    routing_key='high_priority',
    body=task_json,
    properties=pika.BasicProperties(
        delivery_mode=2,  # 持久化
    )
)

# 消费者消费
channel.basic_consume(
    queue='high_priority',
    on_message_callback=callback,
    auto_ack=False  # 手动ACK
)

注意那个 auto_ack=False。我曾经见过一个事故：某团队设了自动 ACK，消费者拿到消息后还没处理就挂了，消息直接丢失。嗯，血的教训。

4.3 Kafka：高吞吐的王者

Kafka 的设计哲学跟 RabbitMQ 完全不同。它不追求「每条消息都确认」，而是追求「海量消息快速流转」。在调度场景里，Kafka 适合：

• 日志采集类任务（每天几亿条）
• 数据同步任务（批量处理）
• 事件驱动型调度（比如用户行为触发任务）

Kafka 的核心概念是 Partition（分区）。每个分区内的消息是有序的，但分区之间不保证顺序。我一般建议：

• 如果任务需要严格顺序，用同一个 Key 路由到同一个分区
• 如果不需要顺序，随机分区就行，吞吐更高

# 生产者示例
producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer.send('task_topic', key=b'user_123', value=task_bytes)

# 消费者示例
consumer = KafkaConsumer(
    'task_topic',
    bootstrap_servers='localhost:9092',
    group_id='task_workers',
    enable_auto_commit=False
)
for msg in consumer:
    process_task(msg.value)
    consumer.commit()  # 手动提交偏移量

你想想看，Kafka 的吞吐为什么高？因为它把消息顺序写到磁盘，利用操作系统的 Page Cache，读的时候直接从内存命中。我测过，单机 Kafka 轻松扛 10 万条/秒。

4.4 延迟队列的实现

延迟队列是调度系统里的「定时炸弹」——任务不是立刻执行，而是等到指定时间才触发。比如：订单 30 分钟未支付自动取消、用户注册 24 小时后发送回访邮件。

实现延迟队列，我见过四种主流方案：

方案一：Redis ZSet（我最常用的）

用 ZSet 的 Score 存执行时间戳，轮询取出到期的任务：

# 添加延迟任务
ZADD delay_queue 1700000000 "{'id': 1001, 'delay': 30}"

# 轮询取出到期任务
ZRANGEBYSCORE delay_queue 0 1700000000 WITHSCORES

优点：实现简单，性能好。缺点：轮询间隔不好控制。我一般设 1 秒轮询一次，精度够用。

方案二：RabbitMQ 死信队列

RabbitMQ 有个特性——消息可以设置 TTL（存活时间）。TTL 到期后，消息自动转入死信队列。利用这个机制：

# 声明主队列，设置TTL和死信交换机
channel.queue_declare(
    queue='delay_queue',
    arguments={
        'x-message-ttl': 30000,  # 30秒
        'x-dead-letter-exchange': 'dlx_exchange',
        'x-dead-letter-routing-key': 'dlx_routing'
    }
)

# 消费者监听死信队列
channel.basic_consume(
    queue='dlx_queue',
    on_message_callback=callback
)

优点：利用 RabbitMQ 原生机制，可靠。缺点：每个延迟时间需要单独一个队列，管理麻烦。

方案三：Kafka 时间轮

Kafka 内部有个时间轮（TimingWheel）算法，用来处理延迟任务。不过 Kafka 本身不直接提供延迟队列 API，需要自己封装。我见过一些团队用 Kafka + 外部时间轮库实现，但说实话，有点重。

方案四：数据库轮询（不推荐）

把任务存到数据库，定时扫表。嗯，我刚开始工作时就这么干过。后来发现，10 万条任务扫一次要 2 秒，延迟根本控不住。除非任务量极小（千级以下），否则别碰。

4.5 知识体系总览

下面这张图，是我梳理的任务队列选型逻辑：

这张图的核心逻辑很简单：先看吞吐量需求，再看是否需要延迟。Redis 适合轻量场景，RabbitMQ 适合可靠场景，Kafka 适合海量场景。延迟队列的实现，Redis ZSet 和 RabbitMQ 死信队列是两种主流选择。

好了，这一章就聊到这儿。记住一句话：没有最好的队列，只有最合适的队列。选型的时候，多想想你的业务场景——任务量多大？可靠性要求多高？延迟精度多少？想清楚了，方案自然就出来了。

想要更多资料，联系可可