⚙️ 异步任务与队列

Redis Queue / BullMQ / Celery / Temporal / Inngest — 当「请求-响应」不够用时，如何构建可靠的异步处理系统

1. 为什么需要异步任务

HTTP 请求-响应模型有一个硬限制：用户不想等。当一个操作需要几秒甚至几分钟——发邮件、生成报表、视频转码、AI 推理——同步处理会让请求超时、连接断开、用户体验灾难。

典型的异步场景：

• 邮件发送：注册确认、密码重置、营销邮件——SMTP 一次调用 1-5 秒
• 文件处理：图片压缩、PDF 生成、视频转码——CPU 密集，秒到小时级
• 第三方 API 调用：支付回调、AI 推理、外部同步——网络不确定，可能重试
• 数据批处理：每日报表、数据导入、索引重建——数据量大，必须离线
• 定时任务：清理过期数据、发送提醒、同步远程数据——周期执行
• Agent 任务：LLM 多轮推理、工具调用链——可能执行几分钟

2. 核心概念

在深入各个工具之前，先理解异步任务系统的通用概念——这些概念在不同工具中以不同名字出现，但本质相同。

2.1 Producer → Queue → Consumer 模型

这是最基本的模型。Producer（生产者）将任务描述序列化后放入 Queue（队列），Consumer（消费者/Worker）从队列取出任务执行。生产者和消费者解耦，可以独立扩展。

2.2 关键术语

2.3 三个层次的"可靠"

3. Redis Queue (Python)

Python 轻量 Redis

RQ (Redis Queue) 是 Python 生态最简单的任务队列——直接用 Redis 的 LPUSH/BRPOP 实现队列，没有额外依赖，5 分钟上手。

3.1 是什么

RQ 是一个极简的 Python 任务队列库，将函数调用序列化为 Job 存入 Redis，Worker 进程从 Redis 取出执行。它的哲学是：简单到不需要学习。

• 任务就是普通 Python 函数——无装饰器、无基类、无 DSL
• 队列就是 Redis List——一个 lpush 入队，一个 brpop 出队
• Worker 就是一个进程——rq worker 启动，fork 执行任务

3.2 何时用

• Python 项目，需要快速加一个后台任务
• 任务逻辑简单，不需要复杂编排（DAG、子任务）
• 已经有 Redis，不想引入新的基础设施
• 团队小，不想学习 Celery 的配置地狱

3.3 优劣对比

3.4 实际代码示例

定义任务（就是普通函数）

# tasks.py
import time
import requests

def send_welcome_email(user_id: int, email: str):
    """发送欢迎邮件——一个普通的 Python 函数"""
    time.sleep(2)  # 模拟 SMTP 调用
    print(f"Welcome email sent to {email}")
    return {"status": "sent", "user_id": user_id}

def generate_report(user_id: int, report_type: str):
    """生成报表——CPU 密集任务"""
    time.sleep(10)  # 模拟计算
    report_url = f"https://cdn.example.com/reports/{user_id}-{report_type}.pdf"
    return {"url": report_url, "type": report_type}

def process_image(image_url: str, operations: list):
    """图片处理管道"""
    # 下载 → 处理 → 上传
    img = requests.get(image_url).content
    for op in operations:
        if op == "resize":
            img = resize_image(img, 800, 600)
        elif op == "watermark":
            img = add_watermark(img, "My SaaS")
    return {"processed": True, "original_url": image_url}

入队（Producer 端）

# app.py — Flask 应用中入队
from redis import Redis
from rq import Queue
from tasks import send_welcome_email, generate_report

redis_conn = Redis(host="localhost", port=6379)
q = Queue("emails", connection=redis_conn)  # 高优先级邮件队列
q_low = Queue("reports", connection=redis_conn)  # 低优先级报表队列

# 基本入队
job = q.enqueue(send_welcome_email, 42, "user@example.com")
print(job.id)  # 'a1b2c3d4-...'

# 入队时指定参数
job = q.enqueue(
    generate_report,
    42,
    "monthly",
    job_timeout="10m",     # 任务超时
    result_ttl=3600,       # 结果保留 1 小时
    failure_ttl=86400,     # 失败记录保留 1 天
)

# 延迟执行
from datetime import datetime, timedelta
job = q.enqueue_in(timedelta(hours=1), send_welcome_email, 42, "user@example.com")

# 查询结果
job = q.enqueue(send_welcome_email, 42, "user@example.com")
# ... 稍后 ...
job.refresh()  # 刷新状态
if job.is_finished:
    print(job.result)  # {'status': 'sent', 'user_id': 42}
elif job.is_failed:
    print(job.exc_info)  # 异常信息

启动 Worker

# 命令行启动 Worker
$ rq worker emails reports  # 监听两个队列

# 或在代码中启动
from rq import Worker
worker = Worker([q, q_low], connection=redis_conn)
worker.work()

# 带参数启动
$ rq worker emails \
    --url redis://localhost:6379 \
    --burst  # 处理完所有任务后退出（适合批处理）

3.5 替代方案

• 需要更强大的 Python 队列 → Celery
• Node.js 项目 → BullMQ
• 需要持久化 Workflow → Temporal

4. BullMQ (Node.js)

Node.js Redis 功能丰富

BullMQ 是 Node.js 生态中最强大的 Redis 队列库。它基于 Bull（v3）重写，支持优先级、延迟、限流、父子任务、进度追踪等高级特性，是 NestJS、Vercel 等项目的底层选择。

4.1 是什么

BullMQ 是一个基于 Redis 的高性能任务队列，专为 Node.js/TypeScript 设计。它将 Redis 用到极致——不是简单的 LPUSH/BRPOP，而是利用 Redis 的 Hash、Sorted Set、List、Stream 等多种数据结构实现优先级队列、延迟队列、限流、事件通知等功能。

4.2 何时用

• Node.js/TypeScript 项目，需要可靠的任务队列
• 需要优先级、延迟、限流、子任务等高级特性
• 需要实时进度追踪和事件通知
• 使用 NestJS（内置 BullMQ 集成）
• SaaS 产品中用户触发的后台操作（导出、通知、批处理）

4.3 优劣对比

4.4 实际代码示例

基本用法

// queue.ts
import { Queue, Worker, Job } from 'bullmq';

// 定义队列
const emailQueue = new Queue('emails', {
  connection: { host: 'localhost', port: 6379 },
});

// 入队
await emailQueue.add('welcome', {
  userId: 42,
  email: 'user@example.com',
  template: 'welcome',
}, {
  attempts: 3,               // 最多重试 3 次
  backoff: {                  // 指数退避
    type: 'exponential',
    delay: 1000,              // 1s, 2s, 4s...
  },
  removeOnComplete: { count: 1000 },  // 只保留最近 1000 个完成记录
  removeOnFail: { age: 7 * 24 * 3600 }, // 7 天后删除失败记录
});

// 延迟任务
await emailQueue.add('trial-ending', data, {
  delay: 3 * 24 * 3600 * 1000,  // 3 天后执行
});

// 优先级任务
await emailQueue.add('vip-notification', data, {
  priority: 1,   // 数字越小优先级越高
});

// 定时任务 (Cron)
await emailQueue.add('daily-digest', data, {
  repeat: {
    pattern: '0 9 * * *',      // 每天早上 9 点
    tz: 'Asia/Shanghai',
  },
});

// Worker
const worker = new Worker('emails', async (job: Job) => {
  const { userId, email, template } = job.data;

  // 更新进度
  await job.updateProgress(10);
  const user = await fetchUser(userId);

  await job.updateProgress(50);
  await sendEmail(email, template, user);

  await job.updateProgress(100);
  return { sent: true, email };
}, {
  connection: { host: 'localhost', port: 6379' },
  concurrency: 5,               // 并发处理 5 个任务
  limiter: {                     // 限流：每秒最多 10 个
    max: 10,
    duration: 1000,
  },
});

// 事件监听
worker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed:`, result);
});

worker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed:`, err.message);
});

worker.on('progress', (job, progress) => {
  console.log(`Job ${job.id} progress: ${progress}%`);
});

Flow（任务编排）

// 用 Flow 编排多步骤任务
import { FlowProducer } from 'bullmq';

const flowProducer = new FlowProducer({
  connection: { host: 'localhost', port: 6379 },
});

// 定义工作流：处理订单 → 生成发票 → 发送邮件
const flow = await flowProducer.add({
  name: 'order-confirmation',
  queueName: 'emails',
  data: { template: 'order-confirmed' },
  children: [
    {
      name: 'generate-invoice',
      queueName: 'billing',
      data: { orderId: 'ORD-123' },
      children: [
        {
          name: 'process-payment',
          queueName: 'payments',
          data: { orderId: 'ORD-123', amount: 99.99 },
        },
      ],
    },
  ],
});

// 执行顺序：process-payment → generate-invoice → order-confirmation
// 父任务等所有子任务完成后才执行

Bull Board 监控

// server.ts
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');

createBullBoard({
  queues: [
    new BullMQAdapter(emailQueue),
    new BullMQAdapter(billingQueue),
  ],
  serverAdapter,
});

app.use('/admin/queues', serverAdapter.getRouter());
// 访问 /admin/queues 即可查看任务状态、重试、删除

4.5 替代方案

• 更轻量 → fastq / bee-queue（但功能远不如 BullMQ）
• Python 项目 → Celery
• 需要持久化 Workflow → Temporal
• Serverless 环境 → Inngest

5. Celery (Python)

Python 重量级 企业级

Celery 是 Python 生态中最老牌、最成熟的分布式任务队列。它支持多种 Broker（Redis、RabbitMQ、SQS）和多种 Result Backend，是 Django 项目的默认选择。

5.1 是什么

Celery 是一个分布式任务队列系统，核心由三部分组成：

• Broker：消息中间件（Redis / RabbitMQ / Amazon SQS），负责任务分发
• Worker：执行任务的进程（支持 prefork / gevent / eventlet 并发模型）
• Result Backend：存储任务结果（Redis / RPC / Database / MongoDB）

与 RQ 不同，Celery 提供了完整的任务编排能力：Chain（链式）、Group（并行）、Chord（并行+回调）、Chunk（分块）。

5.2 何时用

• Python 项目，需要企业级任务队列
• Django 项目（django-celery-results / django-celery-beat 无缝集成）
• 需要多种 Broker 支持（RabbitMQ for 可靠、Redis for 简单、SQS for AWS）
• 需要复杂任务编排（Chain / Group / Chord）
• 需要定时调度（Celery Beat — 内置 Cron）
• 大规模部署（Celery 被Instagram、Mozilla 等大规模使用）

5.3 优劣对比

5.4 实际代码示例

配置

# celery_config.py
broker_url = "redis://localhost:6379/0"
result_backend = "redis://localhost:6379/1"

# 序列化
task_serializer = "json"
result_serializer = "json"
accept_content = ["json"]

# 并发
worker_concurrency = 4           # Worker 进程数
worker_prefetch_multiplier = 4   # 每个 Worker 预取的任务数

# 任务超时
task_soft_time_limit = 300       # 软超时 5 分钟（抛异常）
task_time_limit = 600            # 硬超时 10 分钟（SIGKILL）

# 重试
task_acks_late = True            # 任务完成后才确认（避免 Worker 崩溃丢任务）
task_reject_on_worker_lost = True  # Worker 崩溃时重新入队

# 结果
result_expires = 3600            # 结果 1 小时后过期

# Beat 定时
beat_schedule = {
    "cleanup-expired-sessions": {
        "task": "app.tasks.cleanup_expired_sessions",
        "schedule": crontab(hour=3, minute=0),  # 每天凌晨 3 点
    },
    "send-daily-digest": {
        "task": "app.tasks.send_daily_digest",
        "schedule": crontab(hour=9, minute=0, day_of_week="mon-fri"),
    },
}

定义任务

# tasks.py
from celery import shared_task, chain, group, chord
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def send_email(self, to: str, subject: str, body: str):
    """发送邮件，失败自动重试"""
    try:
        smtp_send(to, subject, body)
    except SMTPException as exc:
        logger.warning(f"Email to {to} failed, retrying: {exc}")
        raise self.retry(exc=exc)  # 触发重试

@shared_task
def process_order(order_id: str):
    """处理订单"""
    order = Order.objects.get(id=order_id)
    # 扣库存、发通知、更新状态
    order.status = "processed"
    order.save()
    return {"order_id": order_id, "status": "processed"}

@shared_task
def generate_invoice(order_id: str):
    """生成发票"""
    invoice = Invoice.create(order_id=order_id)
    return {"invoice_id": invoice.id, "url": invoice.pdf_url}

@shared_task
def notify_customer(order_id: str, invoice_url: str):
    """通知客户"""
    order = Order.objects.get(id=order_id)
    send_email.delay(order.email, "Your Invoice", f"View: {invoice_url}")
    return {"notified": True}

任务编排（Canvas）

# Chain：顺序执行，前一步结果传给下一步
workflow = chain(
    process_order.s("ORD-123"),           # 返回 {order_id, status}
    generate_invoice.s(),                  # 接收 order_id
    notify_customer.s(),                   # 接收 invoice_url
)
result = workflow.apply_async()

# Group：并行执行
from celery import group
batch = group(
    send_email.s("a@example.com", "Hi A", "..."),
    send_email.s("b@example.com", "Hi B", "..."),
    send_email.s("c@example.com", "Hi C", "..."),
)
result = batch.apply_async()
result.get()  # [True, True, True]

# Chord：并行执行 + 回调
# 先并行处理所有订单，全部完成后发送汇总
chord(
    [process_order.s(o.id) for o in orders],
    send_summary_email.s(admin_email)   # 回调收到所有结果
).apply_async()

# Chunk：分块并行
# 1000 个邮件，每块 50 个
process_emails.chunks(
    email_list, 50
).apply_async()

Django 集成

# proj/celery.py
import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "proj.settings")
app = Celery("proj")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()  # 自动发现各 app 的 tasks.py

# proj/__init__.py
from .celery import app as celery_app
__all__ = ("celery_app",)

# myapp/tasks.py — 在 Django app 中定义任务
from celery import shared_task

@shared_task
def my_task():
    ...

5.5 替代方案

• 更简单的 Python 队列 → RQ
• Node.js 项目 → BullMQ
• 需要 Workflow 持久化 → Temporal
• Huey：轻量替代（SQLite/Redis/Greenlet），适合小型项目
• Dramatiq：Celery 的现代替代品，API 更清爽

6. Temporal (多语言)

Durable Execution 企业级 多语言

Temporal 不是传统的任务队列——它是持久化执行引擎（Durable Execution Engine）。代码写的像同步，执行的像异步，状态自动持久化，进程崩溃后自动恢复。这是处理复杂业务流程的终极武器。

6.1 是什么

Temporal 的核心思想：代码即 Workflow。你写普通的同步代码，Temporal 保证它最终一定会执行完成——即使进程崩溃、服务器宕机、网络分区。

6.2 何时用

• 业务流程涉及多个步骤，每步可能失败，需要补偿/回滚
• 流程可能运行很长时间（小时/天/周），需要持久化状态
• 需要人工审批步骤（等待 Signal 后继续）
• 需要流程可观测——每一步的执行历史清晰可查
• 微服务编排——跨服务的业务流程协调
• 替代自己写的"状态机 + 定时轮询"系统

6.3 优劣对比

6.4 实际代码示例

TypeScript SDK：订单处理 Workflow

// workflows.ts
import { proxyActivities, setHandler, condition, sleep, workflowInfo }
  from '@temporalio/workflow';
import type * as activities from './activities';

// Activity 代理——实际执行的操作
const { chargeCreditCard, sendConfirmationEmail, refundCreditCard,
        reserveInventory, releaseInventory, scheduleDelivery } =
  proxyActivities({
    startToCloseTimeout: '30s',
    retry: { maximumAttempts: 3 },
  });

// 订单处理 Workflow
export async function processOrder(order: Order): Promise {
  let paymentConfirmed = false;
  let orderCancelled = false;

  // 监听取消 Signal
  setHandler(cancelOrderSignal, () => { orderCancelled = true; });

  try {
    // Step 1: 预留库存
    await reserveInventory(order.items);

    // Step 2: 等待支付确认（最多 30 分钟）
    const paymentReceived = await condition(
      () => paymentConfirmed,
      '30 minutes'
    );

    if (!paymentReceived || orderCancelled) {
      // 超时或取消 → 释放库存
      await releaseInventory(order.items);
      return 'CANCELLED';
    }

    // Step 3: 扣款
    await chargeCreditCard(order.paymentMethod, order.total);

    // Step 4: 安排发货
    const trackingId = await scheduleDelivery(order.shippingAddress);

    // Step 5: 发送确认邮件
    await sendConfirmationEmail(order.email, order.id, trackingId);

    return trackingId;

  } catch (error) {
    // 补偿事务：释放库存 + 退款
    await releaseInventory(order.items);
    if (paymentConfirmed) {
      await refundCreditCard(order.paymentMethod, order.total);
    }
    throw error;
  }
}

Activities 实现

// activities.ts
import { Context } from '@temporalio/activity';

export async function chargeCreditCard(
  paymentMethod: string, amount: number
): Promise {
  // Activity 可以有副作用——调 API、操作 DB
  const heartbeat = Context.current().heartbeat;
  heartbeat('charging');

  const result = await stripe.charges.create({
    amount: Math.round(amount * 100),
    currency: 'usd',
    source: paymentMethod,
  });

  return result.id;  // charge_id
}

export async function sendConfirmationEmail(
  email: string, orderId: string, trackingId: string
): Promise {
  await mailer.send({
    to: email,
    subject: `Order ${orderId} Confirmed`,
    body: `Tracking: ${trackingId}`,
  });
}

// ... reserveInventory, releaseInventory, refundCreditCard, scheduleDelivery

Python SDK：简单示例

# workflows.py
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy

with workflow.unsafe.imports_passed_through():
    from activities import send_email, process_data, save_result

@workflow.defn
class DataProcessingWorkflow:
    """数据处理 Workflow：处理 → 保存 → 通知"""

    @workflow.run
    async def run(self, data_id: str) -> str:
        retry = RetryPolicy(
            maximum_attempts=3,
            initial_interval=timedelta(seconds=1),
            backoff_coefficient=2.0,
        )

        # 处理数据
        result = await workflow.execute_activity(
            process_data,
            data_id,
            start_to_close_timeout=timedelta(minutes=10),
            retry_policy=retry,
        )

        # 保存结果
        url = await workflow.execute_activity(
            save_result,
            result,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=retry,
        )

        # 发通知
        await workflow.execute_activity(
            send_email,
            f"Data {data_id} processed: {url}",
            start_to_close_timeout=timedelta(seconds=30),
        )

        return url

6.5 替代方案

• 简单任务队列 → Celery / BullMQ
• Serverless Durable Execution → Inngest / AWS Step Functions / Azure Durable Functions
• 轻量级替代 → go-work（Go 简单后台处理）
• 类似产品 → Trigger.dev / Windmill

7. Inngest (Serverless)

Serverless Event-Driven Durable

Inngest 是面向 Serverless 环境的持久化执行平台。它把 Temporal 的 Durable Execution 概念搬到了 Serverless 世界——你写普通的函数，Inngest 处理重试、超时、并发、限流，而且不需要运行任何服务器。

7.1 是什么

Inngest 的核心模型是 Event → Function：

• Event：发生的事情（user.created, order.paid）
• Function：响应 Event 的处理逻辑，可以包含多个 Step
• Step：Function 中的一个操作，每步独立重试、独立超时

与 Temporal 的关键区别：Inngest 的 Function 跑在你的 Serverless 函数里（Vercel、Netlify、AWS Lambda），Inngest 只负责调度和状态管理。你不需要运行 Worker 进程。

传统队列：
  App → Queue → Worker Process → 执行
  需要运行 Worker 进程，需要管理 Worker 生命周期

Inngest：
  App → Inngest Cloud → HTTPS 调用你的 Serverless Function → 执行
  不需要 Worker，不需要管理进程，但需要 HTTPS 端点

7.2 何时用

• Serverless 架构（Vercel / Netlify / AWS Lambda / Cloudflare Workers）
• 不想管理 Worker 进程和基础设施
• 需要 Durable Execution 但不想部署 Temporal Server
• Event-Driven 架构——Event 触发多个 Function
• 需要限流、去重、优先级但不想自己实现

7.3 优劣对比

7.4 实际代码示例

TypeScript SDK：用户注册流程

// inngest/functions.ts
import { inngest } from "./client";

// 用户注册后的多步处理
export const onUserCreated = inngest.createFunction(
  {
    id: "on-user-created",
    name: "User Onboarding Flow",
    retries: 3,                    // 最多重试 3 次
    concurrency: {                 // 同一用户串行执行
      limit: 1,
      key: "event.data.userId",
    },
    throttle: {                    // 限流
      limit: 100,
      period: "1m",
    },
  },
  { event: "user.created" },      // 触发 Event
  async ({ event, step }) => {
    const { userId, email, name } = event.data;

    // Step 1: 发送欢迎邮件
    await step.run("send-welcome-email", async () => {
      await sendEmail(email, "welcome", { name });
      return { sent: true };
    });

    // Step 2: 等待 1 天
    await step.sleep("wait-1-day", "1d");

    // Step 3: 发送引导邮件
    await step.run("send-onboarding-email", async () => {
      const hasLoggedIn = await checkLogin(userId);
      if (!hasLoggedIn) {
        await sendEmail(email, "onboarding-reminder", { name });
      }
      return { sent: !hasLoggedIn };
    });

    // Step 4: 等待 3 天
    await step.sleep("wait-3-days", "3d");

    // Step 5: 检查是否激活
    const isActive = await step.run("check-activation", async () => {
      const user = await getUser(userId);
      return user.lastActiveAt > user.createdAt;
    });

    if (!isActive) {
      await step.run("send-re-engagement", async () => {
        await sendEmail(email, "re-engagement", { name });
      });
    }

    return { userId, onboardingComplete: true };
  }
);

Event 发送

// 从你的应用中发送 Event
import { inngest } from "./client";

// 用户注册时
app.post("/register", async (req, res) => {
  const user = await createUser(req.body);

  // 发送 Event → Inngest 调度 Function
  await inngest.send({
    name: "user.created",
    data: {
      userId: user.id,
      email: user.email,
      name: user.name,
    },
  });

  res.json({ user });  // 立即返回，不等后台处理
});

// 也可以批量发送
await inngest.send([
  { name: "order.created", data: { orderId: "1", ... } },
  { name: "order.created", data: { orderId: "2", ... } },
]);

Cron 定时

// 定时任务
export const dailyCleanup = inngest.createFunction(
  { id: "daily-cleanup", name: "Daily Cleanup" },
  { cron: "0 3 * * *" },  // 每天凌晨 3 点
  async ({ step }) => {
    await step.run("cleanup-expired-sessions", async () => {
      const count = await Session.deleteExpired();
      return { deleted: count };
    });

    await step.run("cleanup-temp-files", async () => {
      const count = await TempFile.deleteOlderThan(7);
      return { deleted: count };
    });
  }
);

Python SDK

# functions.py
from inngest import Inngest, Function, Event, Step

client = Inngest(app_id="my-saas")

@client.create_function(
    fn_id="process-upload",
    trigger=Event(event="file.uploaded"),
    retries=3,
)
async def process_upload(ctx: dict, step: Step):
    file_url = ctx["event"]["data"]["url"]

    # Step 1: 验证文件
    validated = await step.run("validate", async () => {
        return await validate_file(file_url)
    })

    if not validated["ok"]:
        return {"status": "rejected", "reason": validated["reason"]}

    # Step 2: 处理文件
    result = await step.run("process", async () => {
        return await process_file(file_url)
    })

    # Step 3: 通知用户
    await step.run("notify", async () => {
        await send_notification(
            ctx["event"]["data"]["userId"],
            f"File processed: {result['output_url']}"
        )
    })

    return {"status": "done", "output_url": result["output_url"]}

7.5 替代方案

• 自部署 Durable Execution → Temporal
• AWS 原生 → AWS Step Functions（更贵，但 AWS 深度集成）
• 类似产品 → Trigger.dev（TypeScript 优先，自部署可选）
• 传统 Worker 模型 → BullMQ

8. 全维度对比

9. 选型决策树

10. 实战模式

10.1 幂等性：异步任务的基石

所有"至少一次"交付的系统都要求消费者幂等——同一任务执行多次，结果相同。

# ❌ 非幂等：每次执行都创建新记录
def process_payment(order_id):
    order = Order.get(order_id)
    charge = Charge.create(amount=order.total)  # 重试会重复扣款！
    return charge.id

# ✅ 幂等：先检查是否已处理
def process_payment(order_id):
    order = Order.get(order_id)

    # 检查是否已有成功的扣款
    existing = Charge.find_by_order(order_id, status="succeeded")
    if existing:
        return existing.id  # 已处理过，直接返回

    charge = Charge.create(amount=order.total, idempotency_key=order_id)
    return charge.id

# ✅ 更好：用数据库唯一约束保证幂等
def process_payment(order_id):
    try:
        charge = Charge.create(
            amount=Order.get(order_id).total,
            idempotency_key=f"order-{order_id}"  # UNIQUE 约束
        )
    except UniqueViolation:
        # 已经创建过，返回已有的
        charge = Charge.find_by_idempotency_key(f"order-{order_id}")
    return charge.id

10.2 Saga Pattern：分布式事务的补偿方案

微服务中，跨服务的事务不能用数据库事务。Saga Pattern 的思路：每一步都有对应的补偿操作，某步失败时，逆序执行之前步骤的补偿。

# Temporal 实现 Saga Pattern
@workflow.defn
class OrderSagaWorkflow:
    @workflow.run
    async def run(self, order: Order):
        # 补偿栈——失败时逆序执行
        compensations = []

        try:
            # Step 1: 预留库存
            await workflow.execute_activity(reserve_inventory, order.items)
            compensations.append(("release_inventory", order.items))

            # Step 2: 扣款
            charge_id = await workflow.execute_activity(charge_payment, order)
            compensations.append(("refund_payment", charge_id))

            # Step 3: 创建发货单
            shipment_id = await workflow.execute_activity(create_shipment, order)
            compensations.append(("cancel_shipment", shipment_id))

            # 全部成功
            return {"status": "completed", "shipment_id": shipment_id}

        except Exception:
            # 逆序执行补偿
            for comp_name, comp_arg in reversed(compensations):
                try:
                    await workflow.execute_activity(
                        get_activity(comp_name), comp_arg,
                        start_to_close_timeout=timedelta(seconds=30),
                    )
                except Exception as comp_err:
                    # 补偿也失败！→ 人工干预
                    logger.error(f"Compensation failed: {comp_err}")
                    await workflow.execute_activity(
                        alert_human, order.id, comp_name, comp_err
                    )
            raise

10.3 限流与背压

当任务产生速度 > 消费速度时，需要限流和背压机制防止系统崩溃。

// BullMQ: Worker 级限流
const worker = new Worker('api-calls', async job => {
  return await callExternalAPI(job.data);
}, {
  limiter: {
    max: 10,        // 每秒最多 10 个
    duration: 1000,
  },
  concurrency: 5,   // 同时处理 5 个
});

// BullMQ: Queue 级限流（防止队列无限增长）
await queue.add('api-call', data, {
  lifo: false,      // FIFO（先进先出）
});

// 检查队列深度——背压
const counts = await queue.getJobCounts();
if (counts.waiting > 10000) {
  // 队列积压过多，拒绝新任务或降级
  throw new ServiceUnavailableError("Queue overloaded, try again later");
}

10.4 优先级队列

# Celery: 多队列模拟优先级
# 定义多个队列
app.conf.task_queues = {
    'critical': Queue('critical', routing_key='critical'),
    'default':  Queue('default', routing_key='default'),
    'low':      Queue('low', routing_key='low'),
}

# 指定任务队列
@shared_task(queue='critical')
def send_password_reset(email):
    ...  # 密码重置邮件，高优先级

@shared_task(queue='low')
def generate_weekly_report():
    ...  # 周报，低优先级

# Worker 监听多个队列（按顺序消费）
# $ celery -A proj worker -Q critical,default,low
# 会先消费 critical，然后 default，最后 low

10.5 死信队列处理

# Celery: 处理死信
@shared_task(bind=True, max_retries=3)
def risky_task(self, data):
    try:
        return do_something_risky(data)
    except Exception as exc:
        if self.request.retries >= self.max_retries:
            # 重试耗尽 → 转入死信处理
            handle_dead_letter.delay(str(self.request.id), data, str(exc))
            return {"status": "dead_lettered", "error": str(exc)}
        raise self.retry(exc=exc)

@shared_task
def handle_dead_letter(task_id, data, error):
    """死信处理：记录 + 告警"""
    DeadLetterRecord.create(
        task_id=task_id,
        data=data,
        error=error,
        timestamp=datetime.utcnow(),
    )
    # 发 Slack 告警
    slack_notify(f"💀 Task {task_id} dead-lettered: {error}")

11. 常见坑与教训

12. 与 Agent 架构的关联

异步任务队列在 Agent 系统中扮演关键角色——Agent 的执行本身就是异步任务。从 Agent 研究报告中可以看到：

推荐组合

← 返回 SaaS 知识库  |  Agent 研究报告  |  后端框架选型 →  |  缓存策略 →