3. 异步任务基础

在构建高并发、低延迟的缓存一致性系统时，同步阻塞模型往往成为性能瓶颈。异步任务通过将耗时操作（如缓存重建、数据同步）剥离出主请求链路，显著提升系统吞吐量与响应速度。本节将系统阐述异步任务的核心概念、与同步模式的对比，以及主流的异步任务框架选型。

3.1 异步任务的概念

异步任务（Asynchronous Task）是指将任务的提交与执行在时间上解耦的编程模型。调用方仅负责将任务描述（消息或函数调用）投递到中间件（如消息队列、任务队列），随即立即返回，继续处理其他请求；任务的真正执行由后台的工作进程（Worker）在未来的某个时间点完成。

异步任务在缓存一致性场景中的典型价值：

• 写后失效不阻塞：更新数据库后，异步发送缓存失效消息，主线程无需等待缓存操作完成。
• 批量重建防雪崩：缓存穿透时，仅一个任务负责重建，其余请求等待或快速返回旧值。
• 削峰填谷：突发写入高峰时，任务队列充当缓冲区，Worker 按自身节奏消费。

3.2 异步任务与同步任务的区别

同步任务与异步任务在调用行为、资源占用、错误处理等方面存在本质差异。下表从多个维度进行对比：

3.3 常见的异步任务框架

业界主流的异步任务框架包括 Celery、RQ 和基于 Redis Streams 的自研方案。三者均以 Redis 作为核心中间件，但在功能定位、复杂度、运维成本上各有侧重。

3.3.1 Celery

Celery 是 Python 生态中最成熟的分布式任务队列，支持多种 Broker（Redis、RabbitMQ）和 Result Backend。它提供了任务调度、定时任务、工作流、重试机制等企业级特性。

• 优势：功能全面、社区活跃、支持任务优先级与 ETA。
• 劣势：配置复杂、依赖较多、调试成本较高。
• 适用：大型项目、需要复杂任务编排的场景。

# Celery 任务定义示例
from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3)
def invalidate_cache(self, key):
    try:
        # 执行缓存失效逻辑
        redis_client.delete(key)
    except ConnectionError as exc:
        raise self.retry(exc=exc, countdown=60)

3.3.2 RQ (Redis Queue)

RQ 是一个轻量级的 Python 任务队列，直接依赖 Redis 的 List 数据结构。它设计简洁，适合中小型项目或对运维要求较低的场景。

• 优势：极简 API、零配置、易于监控（通过 RQ Dashboard）。
• 劣势：功能有限（无内置定时任务、无任务优先级）。
• 适用：中小型项目、快速原型开发。

# RQ 任务定义示例
from redis import Redis
from rq import Queue

redis_conn = Redis()
queue = Queue(connection=redis_conn)

def rebuild_cache(key):
    data = db.query("SELECT * FROM ...")
    redis_client.setex(key, 3600, data)

# 投递任务
job = queue.enqueue(rebuild_cache, 'user:123')

3.3.3 Redis Streams

Redis 5.0 引入的 Stream 数据结构提供了持久化消息队列能力，支持消费者组、消息确认（ACK）、消息回溯等特性。适合需要高可控性和低依赖的团队自研任务系统。

• 优势：无额外依赖、消息持久化、支持消费者组水平扩展。
• 劣势：需自行实现任务调度、重试逻辑、监控面板。
• 适用：对 Redis 运维熟悉、需要定制化任务系统的团队。

# Redis Streams 生产者/消费者示例（伪代码）
# 生产者
redis.xadd('task:stream', {'key': 'user:456', 'action': 'invalidate'})

# 消费者（Worker）
while True:
    messages = redis.xreadgroup('group1', 'worker1', {'task:stream': '>'}, count=1, block=5000)
    for msg_id, msg_data in messages:
        try:
            process_task(msg_data)
            redis.xack('task:stream', 'group1', msg_id)
        except Exception:
            # 进入死信队列或重试
            redis.xadd('task:dead', msg_data)

3.4 框架选型决策矩阵

下表从多个工程维度给出量化参考，帮助团队根据自身情况做出选择：