4. 异步任务编排基础：Python asyncio 原理、事件循环、协程、Future 与 Task，Celery/RQ 任务队列简介

在构建高吞吐、低延迟的数据库与缓存联合调试系统时，异步编程与任务编排是核心支撑技术。本章将深入剖析 Python asyncio 的底层原理，并对比 Celery 与 RQ 两种主流任务队列的适用场景。

4.1 Python asyncio 核心概念

asyncio 是 Python 用于编写并发代码的标准库，它通过单线程事件循环实现协作式多任务处理，特别适合 I/O 密集型操作（如数据库查询、缓存读写）。

4.1.1 事件循环（Event Loop）

事件循环是 asyncio 的调度核心，它不断检查是否有待处理的协程、回调或 I/O 事件，并决定下一步执行哪个任务。事件循环运行在单线程中，通过 await 挂起当前协程，将控制权交还给循环。

4.1.2 协程（Coroutine）

协程是通过 async def 定义的函数。调用协程函数不会立即执行，而是返回一个协程对象。协程必须被事件循环调度才能运行，通过 await 关键字挂起并等待另一个协程完成。

4.1.3 Future 与 Task

• Future：一个低层级对象，代表异步操作的最终结果。通常不直接使用，而是作为 Task 的底层实现。
• Task：是 Future 的子类，用于将协程包装成一个可被事件循环调度的独立任务。通过 asyncio.create_task() 创建，任务会立即被调度执行。

4.1.3 执行流程对比

4.2 异步任务编排实战模式

在联合调试场景中，常需要编排多个异步任务：先查缓存，缓存未命中则查数据库，再回填缓存。asyncio 提供了多种编排原语。

4.2.1 并发执行：gather 与 wait

import asyncio

async def fetch_from_db(key: str) -> str:
    await asyncio.sleep(0.1)  # 模拟数据库查询
    return f"db_value_{key}"

async def fetch_from_cache(key: str) -> str | None:
    await asyncio.sleep(0.02)  # 模拟缓存查询
    return None  # 模拟缓存未命中

async def main():
    # 并发执行两个查询
    results = await asyncio.gather(
        fetch_from_cache("user:1"),
        fetch_from_db("user:1"),
        return_exceptions=True
    )
    cache_result, db_result = results
    print(f"Cache: {cache_result}, DB: {db_result}")

asyncio.run(main())

4.2.2 超时控制：wait_for

async def query_with_timeout():
    try:
        result = await asyncio.wait_for(
            fetch_from_db("slow_query"),
            timeout=0.5
        )
        return result
    except asyncio.TimeoutError:
        return "fallback_value"

4.2.3 任务取消与清理

async def controlled_task():
    task = asyncio.create_task(fetch_from_db("key"))
    # 如果 0.3 秒未完成则取消
    try:
        result = await asyncio.wait_for(task, timeout=0.3)
    except asyncio.TimeoutError:
        task.cancel()
        try:
            await task  # 等待取消完成
        except asyncio.CancelledError:
            pass
        result = "timeout_fallback"
    return result

4.3 任务队列：Celery 与 RQ

当异步任务需要跨进程、跨机器执行，或需要持久化、重试、定时调度时，需要引入任务队列。Celery 和 RQ 是 Python 生态中最常用的两个选择。

4.3.1 架构对比

4.3.2 Celery 快速示例

# tasks.py
from celery import Celery

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

@app.task(bind=True, max_retries=3)
def sync_cache_to_db(self, cache_key: str, value: str):
    try:
        # 模拟将缓存数据同步到 PostgreSQL
        print(f"Syncing {cache_key} -> {value}")
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

# 调用
sync_cache_to_db.delay("user:1", "new_value")

4.3.3 RQ 快速示例

# worker.py
import redis
from rq import Queue, Connection
from rq.worker import Worker

def sync_cache_to_db(cache_key: str, value: str):
    print(f"RQ Worker syncing {cache_key} -> {value}")

# 启动 Worker
with Connection(redis.Redis()):
    q = Queue()
    Worker(q).work()

# 调用
from rq import Queue
from redis import Redis
q = Queue(connection=Redis())
q.enqueue(sync_cache_to_db, "user:1", "new_value")

4.4 选型建议：asyncio vs 任务队列