Python 与 Celery：分布式任务队列的设计、实现与工程实践

Celery 是 Python 生态中最成熟的分布式任务队列框架，广泛应用于异步任务处理、定时调度、分布式计算等场景。本文从形式化定义出发，系统阐述 Celery 的架构设计（Broker-Worker-Backend 三角模型）、AMQP 协议映射、任务状态机、Canvas 工作流原语（chain/group/chord/map/starmap）、Beat 定时调度、重试与幂等性机制，并结合生产级案例（电商订单异步处理、报表生成、邮件批量发送）分析部署、监控与性能优化实践。

1. 学习目标

本文依据 Bloom’s Taxonomy（布鲁姆认知目标分类学）的六个层次组织学习目标，确保从低阶认知到高阶创造的渐进式掌握。

1.1 记忆（Remembering）

• 列出 Celery 的三大核心组件：Broker、Worker、Result Backend。
• 回忆任务状态的七种取值：PENDING、STARTED、SUCCESS、FAILURE、RETRY、REVOKED、REJECTED。
• 列出 Canvas 工作流的五种原语：chain、group、chord、map、starmap。
• 陈述 Celery 的四种并发池：prefork、eventlet、gevent、solo。

1.2 理解（Understanding）

• 解释 Broker 在 Celery 架构中的作用：消息中间件，解耦生产者与消费者。
• 描述任务从发起到完成的消息流转过程：apply_async → Broker → Worker → Backend。
• 区分 delay() 与 apply_async() 的差异：前者是后者的语法糖。
• 解释 Worker 的预取（prefetch）机制及其对任务调度的影响。

1.3 应用（Applying）

• 使用 @app.task 装饰器定义异步任务。
• 使用 chain、group、chord 组合复杂工作流。
• 配置 Celery Beat 实现定时任务调度。
• 实现任务重试与指数退避策略。

1.4 分析（Analyzing）

• 分析 Celery 与 RabbitMQ、Redis 的协作机制，比较两种 Broker 的优劣。
• 解构 AsyncResult 的状态查询机制，分析其性能瓶颈。
• 比较不同并发池（prefork/eventlet/gevent/solo）在 CPU 密集型与 I/O 密集型任务下的表现。
• 分析任务幂等性的重要性及实现策略。

1.5 评价（Evaluating）

• 评估 Celery 在微服务架构中的适用性，对比 Kafka、Amazon SQS 等替代方案。
• 评判 Celery 的”至少一次”（at-least-once）语义对业务逻辑的影响。
• 评价 Celery 5.x 相对于 4.x 的改进与迁移成本。

1.6 创造（Creating）

• 设计一套支持优先级队列与延迟任务的订单处理系统。
• 构建基于 Celery + Flower + Prometheus 的可观测任务监控平台。
• 实现自定义 Backend 以适配特定存储（如 Elasticsearch）。

2. 历史动机与背景

2.1 Celery 的诞生

2009 年，Ask Solem Hoel 在挪威的 PyCon 上首次发布 Celery。当时他在从事 Web 爬虫与数据处理工作，需要一个可靠的异步任务执行框架。已有的解决方案（如 Python 自带的 threading、multiprocessing）无法满足分布式场景的需求，而商业消息队列（如 RabbitMQ）的 Python 客户端库（如 pika）使用复杂，缺乏任务抽象。

Celery 的设计目标：

1. 简单性：通过 @app.task 装饰器，将普通函数转化为可分布式执行的任务。
2. 可靠性：基于消息队列的”至少一次”投递语义，确保任务不丢失。
3. 可扩展性：支持多 Worker、多机器部署，线性扩展处理能力。
4. 灵活性：支持多种 Broker（RabbitMQ、Redis、Amazon SQS）与 Backend（Redis、数据库、Elasticsearch）。

2.2 演进历程

• Celery 2.x（2010）：首个稳定版本，仅支持 RabbitMQ。
• Celery 3.x（2013）：引入 Redis Broker 支持，增加 Canvas 工作流原语。
• Celery 4.x（2016）：重写底层通信层（Kombu），支持 Python 3，引入 asyncio 兼容。
• Celery 5.x（2020）：全面拥抱 Python 3.6+，简化 CLI（celery -A 替代 celery --app），改进 Canvas 实现，移除过时特性。
• Celery 6.x（规划中）：计划原生支持 asyncio，改进 Canvas 性能。

2.3 应用场景

Celery 在现代 Web 应用中无处不在：

• 异步任务：发送邮件、生成报表、处理视频、PDF 转换。
• 定时任务：日报生成、数据同步、缓存预热。
• 分布式计算：机器学习模型训练、大规模数据处理。
• 流量削峰：秒杀场景下的订单异步处理。
• 实时通知：WebSocket 消息推送、Slack/钉钉通知。

2.4 与其他方案的对比

3. 形式化定义

3.1 Celery 架构形式化

Celery 系统可形式化为五元组：

• $P$：生产者（Producer），发起任务的客户端，调用 task.apply_async() 或 task.delay()。
• $B$：消息代理（Broker），存储任务消息，如 RabbitMQ、Redis。
• $W$：工作进程集合（Workers），$W = \{w_1, w_2, ..., w_n\}$，每个 Worker 从 Broker 消费任务。
• $R$：结果后端（Result Backend），存储任务结果与状态。
• $S$：调度器（Scheduler），Celery Beat 进程，按计划发起任务。

3.2 任务形式化

任务 $T$ 是一个五元组：

• $id$：任务唯一标识（UUID）。
• $name$：任务名，格式为 module.function。
• $args$：位置参数列表。
• $kwargs$：关键字参数字典。
• $options$：任务选项（queue、routing_key、expires、retry 等）。

任务消息是 $T$ 的序列化形式，通过 Broker 传输。

3.3 任务状态机形式化

任务状态 $\sigma$ 的状态转换：

状态转换函数：

合法转换：

3.4 Canvas 工作流形式化

Canvas 是 Celery 的工作流原语集合，用于组合多个任务：

• chain（链）：串行执行，前一个任务的结果作为后一个任务的输入。

• group（组）：并行执行，返回所有结果。

• chord（和弦）：group + callback，所有并行任务完成后执行回调。

• map（映射）：对列表中的每个元素应用同一任务。

• starmap（星映射）：类似 map，但每个元素是参数元组。

3.5 Broker 消息模型形式化

Broker 可形式化为消息队列的集合：

每个队列 $Q_i$ 是 FIFO 消息序列：

Worker $w_j$ 从队列 $Q_i$ 消费消息：

预取（prefetch）机制：Worker 一次从队列获取 $N$ 条消息（$N$ 由 worker_prefetch_multiplier 配置），处理完成后才获取下一批。

4. 理论推导

4.1 AMQP 协议与 Celery 消息路由

Celery 基于 AMQP（Advanced Message Queuing Protocol）协议进行消息路由。AMQP 的核心概念：

• Exchange（交换机）：接收生产者发送的消息，根据路由规则分发到队列。
  • direct：根据 routing_key 精确匹配。
  • topic：根据 routing_key 模式匹配（支持通配符 *、#）。
  • fanout：广播到所有绑定的队列。
  • headers：根据消息头属性匹配。
• Queue（队列）：存储消息，等待消费者消费。
• Binding（绑定）：将 Exchange 与 Queue 关联，指定路由规则。

Celery 的默认路由模型：

Producer → Exchange(celery, direct) → Queue(celery) → Worker

通过 task_routes 配置可实现任务路由：

app.conf.task_routes = {
    'tasks.email.*': {'queue': 'email_queue'},
    'tasks.report.*': {'queue': 'report_queue'},
}

4.2 任务序列化机制

Celery 将任务参数序列化为消息体传输。支持的序列化格式：

安全警告：pickle 反序列化时可执行任意代码，仅在可信环境中使用。生产环境必须使用 json。

4.3 Worker 并发模型分析

Celery 支持四种并发池，适用于不同场景：

4.3.1 prefork（默认，多进程）

• 机制：使用 multiprocessing 创建多个子进程，每个进程独立 Python 解释器。
• 优势：真正并行，绕过 GIL，适合 CPU 密集型任务。
• 劣势：进程创建开销大，内存占用高（每进程约 50-100MB）。
• 配置：--concurrency=4（默认 CPU 核心数）。

4.3.2 eventlet（协程）

• 机制：使用 eventlet 库实现绿色线程（协程），单进程内并发。
• 优势：内存占用低（每协程约 10KB），适合 I/O 密集型任务（网络请求、数据库）。
• 劣势：无法绕过 GIL，不适合 CPU 密集型；需 monkey-patch 标准库。
• 配置：--pool=eventlet --concurrency=1000。

4.3.3 gevent（协程）

• 机制：与 eventlet 类似，使用 gevent 库。
• 优势：与 eventlet 相似，但 API 更稳定。
• 劣势：同 eventlet。
• 配置：--pool=gevent --concurrency=1000。

4.3.4 solo（单线程）

• 机制：单进程单线程，串行执行任务。
• 优势：调试简单，无并发问题。
• 劣势：无并发能力。
• 配置：--pool=solo。

选型建议：

• CPU 密集型：prefork。
• I/O 密集型：eventlet 或 gevent。
• 调试：solo。
• Windows：eventlet 或 solo（prefork 在 Windows 上不可靠）。

4.4 预取机制与公平调度

默认预取配置：worker_prefetch_multiplier = 4，即每个 Worker 子进程预取 4 条消息。

问题：长任务与短任务混合时，预取可能导致不公平调度：

Worker 有 4 个子进程，预取 16 条任务
假设前 16 条都是长任务，后续短任务需等待

解决方案：

• 长任务场景：设置 worker_prefetch_multiplier = 1，避免一个 Worker 占用过多任务。
• 使用优先级队列（RabbitMQ 支持）。
• 使用单独队列分离长短任务。

4.5 至少一次投递与幂等性

Celery 基于”至少一次”（at-least-once）语义：任务至少执行一次，可能执行多次。

多次执行的原因：

• Worker 崩溃后，未确认的消息重新入队。
• 网络抖动导致消息重发。
• 任务超时后被重新调度。

幂等性要求：任务必须设计为幂等的——多次执行的效果与一次相同。

幂等性实现策略：

1. 唯一 ID 检查：任务开始前检查是否已处理过该 ID。
2. 数据库唯一约束：插入重复记录时捕获 IntegrityError。
3. 状态机：业务对象有明确状态，仅在特定状态下执行操作。
4. 去重表：记录已处理的任务 ID，新任务先查询。

5. 代码示例

5.1 基础任务定义与调用

"""
Celery 基础任务定义与调用示例

本模块演示：
1. 创建 Celery 应用
2. 定义异步任务
3. 三种调用方式：delay、apply_async、签名
"""
from celery import Celery, signature
import time

# 创建 Celery 应用
# 参数1: 应用名
# broker: 消息代理地址（Redis）
# backend: 结果后端地址（Redis）
app = Celery(
    'myapp',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)


@app.task
def add(x: int, y: int) -> int:
    """
    简单加法任务
    
    参数:
        x: 第一个加数
        y: 第二个加数
    返回:
        两数之和
    """
    return x + y


@app.task
def send_email(to: str, subject: str, body: str) -> bool:
    """
    模拟发送邮件任务
    
    参数:
        to: 收件人邮箱
        subject: 邮件主题
        body: 邮件正文
    返回:
        是否发送成功
    """
    # 模拟耗时操作
    time.sleep(3)
    print(f"邮件已发送: {to} - {subject}")
    return True


@app.task
def long_running_task(duration: int) -> str:
    """
    长时间运行任务
    
    参数:
        duration: 持续时间（秒）
    返回:
        完成消息
    """
    print(f"开始执行任务，持续 {duration} 秒...")
    time.sleep(duration)
    return f"任务完成，耗时 {duration} 秒"


# 调用方式一：delay（最简单，apply_async 的语法糖）
result = add.delay(4, 6)
print(f"任务 ID: {result.id}")
print(f"任务状态: {result.status}")  # PENDING 或 SUCCESS

# 调用方式二：apply_async（支持更多选项）
result = send_email.apply_async(
    args=['user@example.com', '欢迎注册', '您好，欢迎加入！'],
    # 延迟 10 秒执行
    countdown=10,
    # 任务超时 60 秒
    time_limit=60,
    # 指定队列
    queue='email_queue'
)

# 调用方式三：签名（Signature，不立即执行）
sig = add.signature((4, 6), countdown=5)
# 后续执行
result = sig.apply_async()

# 也可以使用 .s() 简写
sig = add.s(4, 6)
result = sig.apply_async()

5.2 任务结果获取与状态查询

"""
任务结果获取与状态查询示例
"""
from celery.result import AsyncResult
from celery.exceptions import TimeoutError

# 配置 Result Backend
app.conf.result_backend = 'redis://localhost:6379/1'

# 发起任务
result = add.delay(10, 20)

# 检查任务是否完成
print(f"任务完成: {result.ready()}")  # False

# 阻塞等待结果（带超时）
try:
    value = result.get(timeout=10)
    print(f"任务结果: {value}")  # 30
except TimeoutError:
    print("任务执行超时")

# 获取任务状态
print(f"任务状态: {result.status}")  # SUCCESS
print(f"任务成功: {result.successful()}")  # True
print(f"任务失败: {result.failed()}")  # False

# 通过任务 ID 查询结果（跨进程）
task_id = result.id
async_result = AsyncResult(task_id, app=app)
print(f"通过 ID 查询: {async_result.result}")

# 失败任务获取异常信息
@app.task
def failing_task():
    raise ValueError("任务执行失败")

result = failing_task.delay()
try:
    result.get(propagate=True)  # propagate=True 时抛出原始异常
except ValueError as e:
    print(f"捕获异常: {e}")

# 获取 traceback
print(f"Traceback: {result.traceback}")

5.3 任务重试与指数退避

"""
任务重试与指数退避策略示例

适用场景：
- 调用外部 API 可能临时失败
- 数据库连接可能超时
- 网络抖动导致请求失败
"""
from celery import Task
from celery.exceptions import Retry
import requests


@app.task(bind=True, max_retries=5)
def call_external_api(self, url: str, params: dict):
    """
    调用外部 API，失败时自动重试
    
    参数:
        self: Task 实例（bind=True 时自动注入）
        url: API 地址
        params: 请求参数
    返回:
        API 响应数据
    """
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        return response.json()
    except (requests.RequestException, requests.Timeout) as exc:
        # 指数退避：5, 10, 20, 40, 80 秒
        retry_count = self.request.retries
        countdown = 5 * (2 ** retry_count)
        print(f"第 {retry_count + 1} 次重试，{countdown} 秒后执行")
        raise self.retry(exc=exc, countdown=countdown)


class CustomTask(Task):
    """
    自定义任务基类，统一处理重试逻辑
    
    优势：
    - 所有任务共享重试策略
    - 集中管理日志与监控
    - 统一异常处理
    """

    # 最大重试次数
    max_retries = 3
    # 重试间隔（秒）
    default_retry_delay = 60

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        """任务失败时的回调"""
        print(f"任务 {task_id} 失败: {exc}")
        # 发送告警
        send_alert.delay(task_id, str(exc))
        super().on_failure(exc, task_id, args, kwargs, einfo)

    def on_success(self, retval, task_id, args, kwargs):
        """任务成功时的回调"""
        print(f"任务 {task_id} 成功，结果: {retval}")

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        """任务重试时的回调"""
        print(f"任务 {task_id} 第 {self.request.retries} 次重试")


@app.task(base=CustomTask, bind=True)
def process_payment(self, order_id: str, amount: float):
    """
    支付处理任务，使用自定义基类
    
    参数:
        self: Task 实例
        order_id: 订单 ID
        amount: 支付金额
    """
    try:
        # 模拟支付 API 调用
        result = requests.post(
            'https://api.payment.com/charge',
            json={'order_id': order_id, 'amount': amount},
            timeout=30
        )
        result.raise_for_status()
        return result.json()
    except requests.RequestException as exc:
        raise self.retry(exc=exc)


@app.task(bind=True, autoretry_for=(Exception,), retry_backoff=True, retry_backoff_max=300, retry_jitter=True)
def fetch_data(self, url: str):
    """
    使用 autoretry 自动重试
    
    autoretry_for: 触发重试的异常类型
    retry_backoff: True 表示指数退避
    retry_backoff_max: 最大退避时间（秒）
    retry_jitter: 添加随机抖动，避免重试风暴
    """
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

5.4 Canvas 工作流：chain、group、chord

"""
Canvas 工作流原语示例

Canvas 是 Celery 的工作流组合工具：
- chain: 串行执行
- group: 并行执行
- chord: 并行 + 回调
"""
from celery import chain, group, chord


@app.task
def process_step1(data: str) -> str:
    """第一步：数据预处理"""
    print(f"步骤1处理: {data}")
    return f"processed_{data}"


@app.task
def process_step2(data: str) -> str:
    """第二步：数据转换"""
    print(f"步骤2处理: {data}")
    return f"transformed_{data}"


@app.task
def process_step3(data: str) -> str:
    """第三步：数据存储"""
    print(f"步骤3处理: {data}")
    return f"stored_{data}"


@app.task
def aggregate_results(results: list) -> dict:
    """聚合多个任务的结果"""
    print(f"聚合结果: {results}")
    return {
        'count': len(results),
        'results': results
    }


# 1. chain：串行执行
# process_step1 → process_step2 → process_step3
workflow = chain(
    process_step1.s("raw_data"),
    process_step2.s(),
    process_step3.s()
)
result = workflow.apply_async()
print(f"chain 最终结果: {result.get()}")  # stored_transformed_processed_raw_data

# 2. group：并行执行
# 同时处理多个数据
parallel_tasks = group(process_step1.s(item) for item in ['a', 'b', 'c', 'd'])
result = parallel_tasks.apply_async()
# 等待所有任务完成
print(f"group 结果: {result.get()}")  # ['processed_a', 'processed_b', 'processed_c', 'processed_d']

# 3. chord：group + 回调
# 并行处理后聚合
callback = aggregate_results.s()
chord_workflow = chord(
    [process_step1.s(item) for item in ['x', 'y', 'z']],
    callback
)
result = chord_workflow.apply_async()
print(f"chord 结果: {result.get()}")  # {'count': 3, 'results': [...]}

# 4. 嵌套工作流：chain + group
# 先并行处理，再串行处理
complex_workflow = chain(
    group(process_step1.s(item) for item in ['a', 'b']),
    process_step2.s(),
    process_step3.s()
)
result = complex_workflow.apply_async()

# 5. 使用 chunks 分批处理
@app.task
def process_batch(items: list) -> int:
    """处理一批数据"""
    return len(items)

# 将 1000 个数据分成 10 批，每批 100 个
batch_workflow = process_batch.chunks(range(1000), 100)
result = batch_workflow.apply_async()
print(f"分批处理结果: {result.get()}")

5.5 Celery Beat 定时任务调度

"""
Celery Beat 定时任务调度示例

Beat 是 Celery 的定时任务调度器，类似 cron
"""
from celery.schedules import crontab, schedule

# 配置 Beat 调度
app.conf.beat_schedule = {
    # 每 30 秒执行一次
    'cleanup-expired-sessions': {
        'task': 'tasks.cleanup_expired_sessions',
        'schedule': 30.0,
    },
    # 每天凌晨 2 点执行
    'daily-report': {
        'task': 'tasks.generate_daily_report',
        'schedule': crontab(hour=2, minute=0),
        'args': (),
    },
    # 每周一上午 9 点执行
    'weekly-summary': {
        'task': 'tasks.generate_weekly_summary',
        'schedule': crontab(hour=9, minute=0, day_of_week=1),
    },
    # 每月 1 号 0 点执行
    'monthly-billing': {
        'task': 'tasks.process_monthly_billing',
        'schedule': crontab(hour=0, minute=0, day_of_month=1),
    },
    # 每 5 分钟执行一次
    'health-check': {
        'task': 'tasks.health_check',
        'schedule': schedule(run_every=300),
    },
}

# 定义定时任务
@app.task
def cleanup_expired_sessions():
    """清理过期的用户会话"""
    from datetime import datetime, timedelta
    from myapp.models import Session

    cutoff = datetime.now() - timedelta(hours=24)
    deleted = Session.objects.filter(updated_at__lt=cutoff).delete()
    print(f"清理了 {deleted[0]} 条过期会话")


@app.task
def generate_daily_report():
    """生成日报"""
    from myapp.services import ReportService
    report = ReportService.generate_daily()
    # 发送邮件
    send_email.delay(
        to='admin@example.com',
        subject=f'日报 - {datetime.now().strftime("%Y-%m-%d")}',
        body=report.summary
    )


@app.task
def generate_weekly_summary():
    """生成周报"""
    from myapp.services import ReportService
    summary = ReportService.generate_weekly()
    send_email.delay(
        to='team@example.com',
        subject='周报',
        body=summary
    )


@app.task
def process_monthly_billing():
    """处理月度账单"""
    from myapp.services import BillingService
    BillingService.process_all_users()


@app.task
def health_check():
    """健康检查"""
    import requests
    try:
        response = requests.get('https://api.example.com/health', timeout=5)
        if response.status_code != 200:
            send_alert.delay('API 健康检查失败')
    except requests.RequestException:
        send_alert.delay('API 不可达')


# 启动 Beat 调度器
# 命令: celery -A tasks beat --loglevel=info
# 或与 Worker 一起启动: celery -A tasks worker --beat --loglevel=info

5.6 任务路由与优先级队列

"""
任务路由与优先级队列示例

适用场景：
- 不同任务类型分配到不同队列
- 重要任务优先处理
- 长短任务分离，避免相互阻塞
"""

# 配置任务路由
app.conf.task_routes = {
    # 邮件相关任务路由到 email_queue
    'tasks.send_email': {'queue': 'email_queue'},
    'tasks.send_bulk_emails': {'queue': 'email_queue'},
    # 报表相关任务路由到 report_queue
    'tasks.generate_report': {'queue': 'report_queue'},
    'tasks.generate_daily_report': {'queue': 'report_queue'},
    # 默认路由到 celery 队列
}

# 定义多个队列
app.conf.task_queues = {
    'celery': {
        'routing_key': 'celery',
    },
    'email_queue': {
        'routing_key': 'email',
    },
    'report_queue': {
        'routing_key': 'report',
    },
    # 高优先级队列
    'high_priority': {
        'routing_key': 'high',
    },
    # 低优先级队列
    'low_priority': {
        'routing_key': 'low',
    },
}

# 默认队列
app.conf.task_default_queue = 'celery'
app.conf.task_default_routing_key = 'celery'


@app.task(queue='high_priority')
def process_urgent_payment(order_id: str):
    """紧急支付任务，放入高优先级队列"""
    print(f"处理紧急支付: {order_id}")


@app.task(queue='low_priority')
def cleanup_temp_files():
    """清理临时文件，放入低优先级队列"""
    import os
    import time
    cutoff = time.time() - 3600  # 1 小时前
    for filename in os.listdir('/tmp'):
        filepath = os.path.join('/tmp', filename)
        if os.path.getmtime(filepath) < cutoff:
            os.remove(filepath)


# 启动 Worker 时指定消费的队列
# celery -A tasks worker -Q celery,email_queue --loglevel=info
# celery -A tasks worker -Q report_queue --loglevel=info
# celery -A tasks worker -Q high_priority --concurrency=2 --loglevel=info


# Redis 支持优先级队列（Celery 5.0+）
app.conf.broker_queue_order = ['high_priority', 'celery', 'low_priority']

# RabbitMQ 支持优先级队列
app.conf.task_queues = {
    'priority_queue': {
        'exchange': 'priority',
        'routing_key': 'priority',
        'queue_arguments': {
            'x-max-priority': 10,  # 0-10，数值越大优先级越高
        },
    },
}

# 发送任务时指定优先级
result = process_urgent_payment.apply_async(
    args=['order_123'],
    priority=9  # 高优先级
)

5.7 任务撤销与超时控制

"""
任务撤销与超时控制示例
"""
from celery.exceptions import SoftTimeLimitExceeded, TimeLimitExceeded


# 撤销任务
result = long_running_task.delay(60)

# 撤销任务（等待中的任务不会执行）
result.revoke()

# 撤销正在执行的任务（发送 SIGTERM 信号）
result.revoke(terminate=True)

# 撤销并等待任务终止
result.revoke(terminate=True, signal='SIGKILL')

# 撤销所有等待中的任务
app.control.purge()


@app.task(time_limit=60, soft_time_limit=50)
def long_task_with_timeout():
    """
    带超时限制的任务
    
    time_limit: 硬超时，超时后 Worker 强制终止任务（SIGKILL）
    soft_time_limit: 软超时，超时后抛出 SoftTimeLimitExceeded，任务可捕获处理
    """
    try:
        # 模拟长时间任务
        for i in range(100):
            # 检查是否超时
            # ... 执行任务逻辑
            import time
            time.sleep(1)
    except SoftTimeLimitExceeded:
        # 软超时，可保存中间状态
        print("任务即将超时，保存中间状态...")
        save_intermediate_state()
        raise  # 重新抛出，触发重试或失败
    except TimeLimitExceeded:
        # 硬超时，无法捕获（进程已被杀死）
        pass


def save_intermediate_state():
    """保存中间状态，便于后续恢复"""
    pass


# Worker 级别的超时配置
app.conf.task_time_limit = 300  # 全局硬超时 5 分钟
app.conf.task_soft_time_limit = 270  # 全局软超时 4.5 分钟


# 任务级超时（apply_async 时指定）
result = long_running_task.apply_async(
    args=[60],
    time_limit=120,  # 2 分钟硬超时
    soft_time_limit=100  # 100 秒软超时
)

5.8 任务进度追踪

"""
任务进度追踪示例

适用场景：
- 长时间任务需要显示进度条
- 批量处理任务需要显示完成数量
- 用户需要知道任务执行状态
"""
from celery import current_task


@app.task(bind=True)
def process_large_dataset(self, dataset_id: str, total_items: int):
    """
    处理大数据集，实时报告进度
    
    参数:
        self: Task 实例
        dataset_id: 数据集 ID
        total_items: 总条目数
    返回:
        处理结果统计
    """
    processed = 0
    failed = 0

    for i in range(total_items):
        try:
            # 模拟处理单个条目
            process_single_item(dataset_id, i)
            processed += 1
        except Exception as e:
            failed += 1
            print(f"处理条目 {i} 失败: {e}")

        # 每 10% 报告一次进度
        if (i + 1) % (total_items // 10) == 0:
            progress = (i + 1) / total_items * 100
            current_task.update_state(
                state='PROGRESS',
                meta={
                    'current': i + 1,
                    'total': total_items,
                    'progress': round(progress, 2),
                    'processed': processed,
                    'failed': failed
                }
            )

    return {
        'status': 'completed',
        'processed': processed,
        'failed': failed,
        'total': total_items
    }


def process_single_item(dataset_id: str, index: int):
    """处理单个数据条目"""
    import random
    if random.random() < 0.05:  # 5% 失败率
        raise ValueError(f"处理失败: {index}")
    # 实际处理逻辑
    pass


# 查询任务进度
result = process_large_dataset.delay('dataset_123', 1000)

# 轮询进度
import time
while not result.ready():
    if result.state == 'PROGRESS':
        info = result.info
        print(f"进度: {info['progress']}% ({info['current']}/{info['total']})")
        print(f"成功: {info['processed']}, 失败: {info['failed']}")
    time.sleep(1)

print(f"最终结果: {result.get()}")


# 使用 Redis 自定义进度追踪（更高效）
import redis

@app.task(bind=True)
def batch_process_with_redis_progress(self, items: list, batch_id: str):
    """使用 Redis 追踪进度（避免频繁更新 Backend）"""
    r = redis.Redis.from_url('redis://localhost:6379/2')
    total = len(items)
    r.hset(f'batch:{batch_id}', 'total', total)
    r.hset(f'batch:{batch_id}', 'processed', 0)

    for i, item in enumerate(items):
        process_item(item)
        r.hincrby(f'batch:{batch_id}', 'processed', 1)

    r.hset(f'batch:{batch_id}', 'status', 'completed')
    r.expire(f'batch:{batch_id}', 3600)  # 1 小时后过期

    return {'batch_id': batch_id, 'total': total}


def get_batch_progress(batch_id: str) -> dict:
    """查询批次进度"""
    r = redis.Redis.from_url('redis://localhost:6379/2')
    data = r.hgetall(f'batch:{batch_id}')
    return {
        'total': int(data.get(b'total', 0)),
        'processed': int(data.get(b'processed', 0)),
        'status': data.get(b'status', b'pending').decode()
    }

6. 对比分析

6.1 Celery vs RQ vs Dramatiq

讨论：Celery 是功能最全面的方案，适合复杂业务场景。RQ 适合轻量级需求，依赖少。Dramatiq 是 Celery 的现代替代品，API 更简洁，但生态较小。

6.2 RabbitMQ vs Redis 作为 Broker

讨论：RabbitMQ 是生产环境 Broker 的首选，提供更强的可靠性保证。Redis 适合开发测试或对可靠性要求不高的场景。

6.3 并发池对比

讨论：CPU 密集型任务用 prefork；I/O 密集型任务用 eventlet 或 gevent，可达到数千并发；调试用 solo；Windows 环境避免用 prefork。

6.4 序列化格式对比

讨论：生产环境必须使用 json，避免 pickle 的安全风险。msgpack 性能最优，但生态较小。yaml 适合配置数据，不适合任务参数。

7. 常见陷阱与反模式

7.1 传递不可序列化的对象

反模式：

@app.task
def process_user(user: User):
    # 错误：User 对象无法 JSON 序列化
    send_email(user.email)

user = User.objects.get(id=1)
process_user.delay(user)  # SerializationError

正确做法：传递 ID，在任务内查询对象。

@app.task
def process_user(user_id: int):
    user = User.objects.get(id=user_id)
    send_email(user.email)

process_user.delay(user.id)

原因：任务参数需经序列化后通过 Broker 传输。json 序列化仅支持基本类型，ORM 对象无法直接序列化。

7.2 在任务中持有数据库连接

事故案例：某 Worker 长期运行后，数据库连接数耗尽，导致新连接失败。

原因：Worker 是长驻进程，数据库连接不会自动释放。每次任务中创建新连接但不关闭，导致连接泄漏。

修复方案：

@app.task
def process_data():
    # 错误：使用全局连接
    # cursor = db.connection.cursor()

    # 正确：任务结束时关闭连接
    from django.db import connection
    try:
        cursor = connection.cursor()
        cursor.execute("SELECT ...")
        return cursor.fetchall()
    finally:
        connection.close()  # 显式关闭连接

# 或配置 Django 自动关闭连接
# DATABASES = {
#     'default': {
#         'CONN_MAX_AGE': 0,  # 每次请求后关闭连接
#     }
# }

7.3 任务幂等性缺失导致重复扣款

事故案例：某支付系统任务重试时重复扣款，导致用户投诉。

原因：任务未设计幂等性，Worker 崩溃后重试时重复执行扣款逻辑。

修复方案：

@app.task(bind=True)
def process_payment(self, order_id: str, amount: float):
    # 检查是否已处理
    if Payment.objects.filter(order_id=order_id).exists():
        return {'status': 'already_processed'}

    try:
        # 使用数据库唯一约束防止重复
        Payment.objects.create(
            order_id=order_id,
            amount=amount,
            status='completed'
        )
    except IntegrityError:
        # 并发场景下，另一个 Worker 已处理
        return {'status': 'already_processed'}

7.4 预取过多导致任务饥饿

事故案例：某系统长任务（5 分钟）与短任务（1 秒）混合，长任务被一个 Worker 全部预取，短任务等待超过 10 分钟。

原因：默认 worker_prefetch_multiplier=4，一个 4 进程的 Worker 预取 16 条任务。若前 16 条都是长任务，后续短任务无法被其他 Worker 消费。

修复方案：

# 方案1：降低预取数
app.conf.worker_prefetch_multiplier = 1

# 方案2：长短任务分队列
# 长任务队列: celery -A tasks worker -Q long_tasks --concurrency=2
# 短任务队列: celery -A tasks worker -Q short_tasks --concurrency=8

# 方案3：使用 task_annotations 动态路由
app.conf.task_annotations = {
    'tasks.long_*': {'queue': 'long_tasks'},
    'tasks.short_*': {'queue': 'short_tasks'},
}

7.5 Windows 上使用 prefork 池崩溃

反模式：

# Windows 上默认使用 prefork，会崩溃
celery -A tasks worker --loglevel=info

原因：Windows 不支持 os.fork()，prefork 池在 Windows 上不可靠。

修复方案：

# 方案1：使用 eventlet
pip install eventlet
celery -A tasks worker --pool=eventlet --loglevel=info

# 方案2：使用 gevent
pip install gevent
celery -A tasks worker --pool=gevent --loglevel=info

# 方案3：使用 solo（无并发）
celery -A tasks worker --pool=solo --loglevel=info

7.6 Result Backend 性能瓶颈

事故案例：高并发场景下（1000 TPS），Result Backend（Redis）成为瓶颈，任务结果写入延迟超过 10 秒。

原因：每次任务完成都会向 Backend 写入状态，高并发下 Redis 写入压力大。

修复方案：

# 方案1：禁用结果存储（不需要结果时）
@app.task(ignore_result=True)
def send_email(to, subject, body):
    ...

# 方案2：使用 result_extended 配置减少写入
app.conf.result_extended = False

# 方案3：使用 RPC Backend（结果直接返回给调用者）
app.conf.result_backend = 'rpc://'

# 方案4：批量写入（Celery 5.1+）
app.conf.result_backend_transport_options = {
    'result_backend_thread_safe': True,
}

7.7 任务循环依赖导致死锁

反模式：

@app.task
def task_a():
    # 任务 A 调用任务 B
    task_b.delay()

@app.task
def task_b():
    # 任务 B 调用任务 A
    task_a.delay()

后果：任务无限循环，Broker 消息堆积，最终耗尽内存。

修复方案：避免任务相互调用，使用 chain 或状态机管理流程。

7.8 Beat 单点故障

事故案例：Beat 进程崩溃，所有定时任务停止执行。

原因：Beat 默认单实例运行，无高可用机制。

修复方案：

# 方案1：使用 celery-redbeat（Redis 分布式锁）
# pip install celery-redbeat
app.conf.beat_scheduler = 'redbeat.RedBeatScheduler'
app.conf.redbeat_redis_url = 'redis://localhost:6379/3'

# 方案2：部署多个 Beat，使用分布式锁
# 通过外部锁确保只有一个 Beat 实例运行

# 方案3：使用 Kubernetes 部署 Beat，配合健康检查与自动重启

8. 工程实践

8.1 生产级 Celery 配置

"""
生产级 Celery 配置

涵盖：
- 序列化与安全
- 性能调优
- 可靠性保证
- 监控与日志
"""
from celery import Celery
from kombu import Queue

app = Celery('myapp')

# 基础配置
app.conf.update(
    # Broker 配置
    broker_url='redis://localhost:6379/0',
    broker_connection_retry_on_startup=True,

    # Result Backend 配置
    result_backend='redis://localhost:6379/1',
    result_serializer='json',
    result_expires=3600,  # 结果 1 小时后过期

    # 序列化配置
    task_serializer='json',
    accept_content=['json'],
    result_accept_content=['json'],

    # 时区配置
    timezone='Asia/Shanghai',
    enable_utc=True,

    # 任务配置
    task_track_started=True,  # 任务开始时记录 STARTED 状态
    task_time_limit=300,  # 硬超时 5 分钟
    task_soft_time_limit=270,  # 软超时 4.5 分钟
    task_acks_late=True,  # 任务完成后才确认消息（避免 Worker 崩溃丢任务）
    task_reject_on_worker_lost=True,  # Worker 丢失时拒绝消息，重新入队

    # Worker 配置
    worker_prefetch_multiplier=1,  # 预取数（长任务场景设为 1）
    worker_max_tasks_per_child=1000,  # 每个 Worker 子进程处理 1000 任务后重启（避免内存泄漏）
    worker_lost_wait=10,  # Worker 丢失后等待 10 秒再重新分配任务

    # 队列配置
    task_queues=(
        Queue('default', routing_key='default'),
        Queue('high_priority', routing_key='high'),
        Queue('low_priority', routing_key='low'),
        Queue('email', routing_key='email'),
        Queue('report', routing_key='report'),
    ),
    task_default_queue='default',
    task_default_routing_key='default',

    # 重试配置
    task_default_retry_delay=60,  # 默认重试间隔 60 秒
    task_default_max_retries=3,  # 默认最大重试 3 次

    # 日志配置
    worker_hijack_root_logger=False,
    worker_log_format='[%(asctime)s: %(levelname)s/%(processName)s] %(message)s',
    worker_task_log_format='[%(asctime)s: %(levelname)s/%(processName)s][%(task_name)s(%(task_id)s)] %(message)s',
)


# 任务路由
app.conf.task_routes = {
    'tasks.email.*': {'queue': 'email'},
    'tasks.report.*': {'queue': 'report'},
    'tasks.urgent_*': {'queue': 'high_priority'},
    'tasks.cleanup_*': {'queue': 'low_priority'},
}


# 任务注解（动态配置）
app.conf.task_annotations = {
    'tasks.long_running_*': {
        'time_limit': 3600,  # 长任务 1 小时超时
        'soft_time_limit': 3500,
    },
    'tasks.quick_*': {
        'time_limit': 30,
        'soft_time_limit': 25,
    },
}

8.2 Django 集成

"""
Django 项目中集成 Celery

项目结构:
myproject/
├── myproject/
│   ├── __init__.py
│   ├── celery.py      # Celery 应用配置
│   ├── settings.py    # Django 配置
├── apps/
│   ├── tasks.py       # 任务定义
"""
# myproject/celery.py
import os
from celery import Celery

# 设置 Django 配置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')

# 从 Django settings 读取配置（以 CELERY_ 为前缀）
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自动发现各 app 的 tasks.py
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    """调试任务，打印请求信息"""
    print(f'Request: {self.request!r}')


# myproject/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)


# myproject/settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 300
CELERY_WORKER_PREFETCH_MULTIPLIER = 1


# apps/tasks.py
from myproject.celery import app
from django.core.mail import send_mail
from django.contrib.auth import get_user_model

User = get_user_model()


@app.task
def send_welcome_email(user_id: int):
    """发送欢迎邮件"""
    user = User.objects.get(id=user_id)
    send_mail(
        subject='欢迎注册',
        message=f'您好 {user.username}，欢迎加入我们！',
        from_email='noreply@example.com',
        recipient_list=[user.email],
    )


@app.task
def generate_user_report(user_id: int):
    """生成用户报告"""
    user = User.objects.get(id=user_id)
    # 生成报告逻辑
    report = f"用户 {user.username} 的报告"
    return {'user_id': user_id, 'report': report}

8.3 监控与可观测性

"""
Celery 监控与可观测性配置

工具：
- Flower: 实时 Web 监控
- Prometheus: 指标采集
- Sentry: 错误追踪
- ELK: 日志聚合
"""
import logging
from celery import signals

# 1. 配置日志
logger = logging.getLogger('celery')


@signals.setup_logging.connect
def setup_logging(**kwargs):
    """自定义日志配置"""
    from logging.config import dictConfig
    dictConfig({
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': {
            'detailed': {
                'format': '%(asctime)s %(levelname)s %(name)s %(process)d %(message)s'
            },
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'formatter': 'detailed',
            },
            'file': {
                'class': 'logging.handlers.RotatingFileHandler',
                'filename': '/var/log/celery/worker.log',
                'maxBytes': 1024 * 1024 * 100,  # 100MB
                'backupCount': 5,
                'formatter': 'detailed',
            },
        },
        'loggers': {
            'celery': {
                'handlers': ['console', 'file'],
                'level': 'INFO',
            },
        },
    })


# 2. 任务执行信号钩子
@signals.task_prerun.connect
def task_prerun_handler(task_id, task, args, kwargs, **extra):
    """任务开始前记录"""
    logger.info(f'任务开始: {task.name} (id={task_id})')


@signals.task_postrun.connect
def task_postrun_handler(task_id, task, args, kwargs, retval, state, **extra):
    """任务结束后记录"""
    logger.info(f'任务结束: {task.name} (id={task_id}, state={state})')


@signals.task_failure.connect
def task_failure_handler(task_id, exception, args, kwargs, traceback, einfo, **extra):
    """任务失败时发送告警"""
    logger.error(f'任务失败: {task_id}, 异常: {exception}')
    # 发送到 Sentry
    import sentry_sdk
    sentry_sdk.capture_exception(exception)


# 3. Prometheus 指标（使用 celery-exporter）
# 安装: pip install celery-exporter
# 启动: celery-exporter --broker-url=redis://localhost:6379/0
# Prometheus 自动采集任务执行数、成功率、延迟等指标


# 4. Flower 监控
# 安装: pip install flower
# 启动: celery -A tasks flower --port=5555
# 访问: http://localhost:5555
# 功能: 实时查看任务状态、Worker 状态、队列长度、任务历史

8.4 性能优化策略

"""
Celery 性能优化策略

1. 批量处理：合并小任务
2. 连接池：复用数据库/Redis 连接
3. 异步 I/O：使用 eventlet/gevent
4. 任务分片：分布式处理大数据
"""

# 1. 批量处理：合并小任务
@app.task
def send_bulk_emails(user_ids: list):
    """批量发送邮件，避免逐个发送"""
    users = User.objects.filter(id__in=user_ids)
    for user in users:
        send_mail(
            subject='通知',
            message='您好，这是一条通知',
            from_email='noreply@example.com',
            recipient_list=[user.email],
        )

# 调用时分批
def trigger_bulk_emails():
    user_ids = list(User.objects.values_list('id', flat=True))
    # 每批 100 个
    batch_size = 100
    for i in range(0, len(user_ids), batch_size):
        batch = user_ids[i:i+batch_size]
        send_bulk_emails.delay(batch)


# 2. 连接池：复用数据库连接
from django.db import connection

@app.task
def query_database():
    """使用连接池查询数据库"""
    with connection.cursor() as cursor:
        cursor.execute("SELECT COUNT(*) FROM auth_user")
        return cursor.fetchone()[0]


# 3. 异步 I/O：eventlet 并发
# 启动: celery -A tasks worker --pool=eventlet --concurrency=1000
@app.task
def fetch_url(url: str):
    """I/O 密集型任务，eventlet 高并发"""
    import requests
    response = requests.get(url, timeout=10)
    return response.status_code


# 4. 任务分片：分布式处理大数据
@app.task
def process_chunk(data_chunk: list):
    """处理数据分片"""
    return [process_item(item) for item in data_chunk]


def process_large_data(data: list, chunk_size: int = 1000):
    """将大数据分片，并行处理"""
    chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]
    # 使用 group 并行处理所有分片
    from celery import group
    job = group(process_chunk.s(chunk) for chunk in chunks)
    result = job.apply_async()
    # 等待所有分片完成
    results = result.get()
    # 合并结果
    return [item for chunk in results for item in chunk]


def process_item(item):
    """处理单个数据项"""
    return item * 2

8.5 高可用部署

# docker-compose.yml: 生产级 Celery 部署
version: '3.8'

services:
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes
    volumes:
      - redis-data:/data
    restart: always

  rabbitmq:
    image: rabbitmq:3.12-management
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: ${RABBITMQ_PASSWORD}
    volumes:
      - rabbitmq-data:/var/lib/rabbitmq
    restart: always

  worker-default:
    build: .
    command: celery -A myapp worker -Q default --concurrency=4 --loglevel=info
    environment:
      - CELERY_BROKER_URL=amqp://admin:${RABBITMQ_PASSWORD}@rabbitmq:5672//
      - CELERY_RESULT_BACKEND=redis://redis:6379/1
      - DATABASE_URL=postgresql://user:pass@db:5432/myapp
    depends_on:
      - rabbitmq
      - redis
    deploy:
      replicas: 3
    restart: always

  worker-high-priority:
    build: .
    command: celery -A myapp worker -Q high_priority --concurrency=8 --loglevel=info
    environment:
      - CELERY_BROKER_URL=amqp://admin:${RABBITMQ_PASSWORD}@rabbitmq:5672//
      - CELERY_RESULT_BACKEND=redis://redis:6379/1
    depends_on:
      - rabbitmq
      - redis
    deploy:
      replicas: 2
    restart: always

  beat:
    build: .
    command: celery -A myapp beat --loglevel=info
    environment:
      - CELERY_BROKER_URL=amqp://admin:${RABBITMQ_PASSWORD}@rabbitmq:5672//
      - CELERY_RESULT_BACKEND=redis://redis:6379/1
    depends_on:
      - rabbitmq
    restart: always

  flower:
    image: mher/flower
    command: celery --broker=amqp://admin:${RABBITMQ_PASSWORD}@rabbitmq:5672// flower --port=5555
    ports:
      - "5555:5555"
    depends_on:
      - rabbitmq
    restart: always

volumes:
  redis-data:
  rabbitmq-data:

9. 案例研究

9.1 案例一：电商订单异步处理

"""
电商订单异步处理系统

流程：
1. 用户下单 → 创建订单任务
2. 扣减库存 → 支付处理 → 发送邮件 → 生成物流单
3. 使用 chain 组合多个任务
"""
from celery import chain


@app.task(queue='high_priority')
def create_order(user_id: int, items: list) -> dict:
    """创建订单"""
    order = Order.objects.create(user_id=user_id, total=calculate_total(items))
    for item in items:
        OrderItem.objects.create(order=order, **item)
    return {'order_id': order.id, 'items': items}


@app.task(queue='high_priority')
def deduct_stock(order_data: dict) -> dict:
    """扣减库存"""
    order_id = order_data['order_id']
    for item in order_data['items']:
        product = Product.objects.get(id=item['product_id'])
        if product.stock < item['quantity']:
            raise ValueError(f"库存不足: {product.name}")
        product.stock -= item['quantity']
        product.save()
    return order_data


@app.task(queue='high_priority')
def process_payment(order_data: dict) -> dict:
    """处理支付"""
    # 幂等性检查
    if Payment.objects.filter(order_id=order_data['order_id']).exists():
        return order_data
    Payment.objects.create(order_id=order_data['order_id'], amount=order_data['total'])
    return order_data


@app.task(queue='email')
def send_order_confirmation(order_data: dict) -> dict:
    """发送订单确认邮件"""
    user = User.objects.get(id=order_data['user_id'])
    send_mail(
        subject=f'订单确认 #{order_data["order_id"]}',
        message='您的订单已确认',
        from_email='orders@example.com',
        recipient_list=[user.email],
    )
    return order_data


@app.task(queue='low_priority')
def generate_shipping_label(order_data: dict) -> dict:
    """生成物流单"""
    shipping = Shipping.objects.create(order_id=order_data['order_id'])
    order_data['tracking_number'] = shipping.tracking_number
    return order_data


# 使用 chain 组合完整流程
def place_order(user_id: int, items: list):
    """下单入口"""
    workflow = chain(
        create_order.s(user_id, items),
        deduct_stock.s(),
        process_payment.s(),
        send_order_confirmation.s(),
        generate_shipping_label.s()
    )
    result = workflow.apply_async()
    return result.id


# 优化：并行化独立步骤
from celery import group, chord

def place_order_optimized(user_id: int, items: list):
    """优化版：支付与邮件并行"""
    workflow = chain(
        create_order.s(user_id, items),
        deduct_stock.s(),
        # 支付后并行执行：发送邮件 + 生成物流单
        process_payment.s(),
        group(send_order_confirmation.s(), generate_shipping_label.s())
    )
    return workflow.apply_async().id

9.2 案例二：批量报表生成

"""
批量报表生成系统

需求：每月为 10000 个用户生成个性化报表

策略：
1. 使用 group 并行生成
2. 使用 chord 聚合结果
3. 分批避免内存爆炸
"""
from celery import group, chord
import pandas as pd


@app.task(queue='report')
def generate_user_report(user_id: int, year: int, month: int) -> dict:
    """生成单个用户的月度报表"""
    user = User.objects.get(id=user_id)
    orders = Order.objects.filter(user_id=user_id, created_at__year=year, created_at__month=month)

    df = pd.DataFrame(list(orders.values('id', 'total', 'created_at')))
    report = {
        'user_id': user_id,
        'username': user.username,
        'total_orders': len(df),
        'total_amount': float(df['total'].sum()) if not df.empty else 0,
        'avg_order_value': float(df['total'].mean()) if not df.empty else 0,
    }

    # 保存报表到文件
    filename = f'reports/{user_id}_{year}{month:02d}.csv'
    df.to_csv(filename, index=False)
    report['file'] = filename
    return report


@app.task(queue='report')
def aggregate_monthly_reports(reports: list) -> dict:
    """聚合所有用户报表，生成总览"""
    df = pd.DataFrame(reports)
    summary = {
        'total_users': len(df),
        'total_revenue': float(df['total_amount'].sum()),
        'avg_revenue_per_user': float(df['total_amount'].mean()),
        'top_users': df.nlargest(10, 'total_amount')[['username', 'total_amount']].to_dict('records'),
    }
    # 发送汇总邮件给管理员
    send_mail(
        subject=f'{reports[0]["year"] if reports else ""}月报表汇总',
        message=str(summary),
        from_email='reports@example.com',
        recipient_list=['admin@example.com'],
    )
    return summary


def generate_monthly_reports(year: int, month: int):
    """触发全量报表生成"""
    user_ids = list(User.objects.values_list('id', flat=True))

    # 分批处理，每批 100 个用户
    batch_size = 100
    for i in range(0, len(user_ids), batch_size):
        batch = user_ids[i:i+batch_size]
        # 使用 chord: 并行生成 + 聚合
        chord(
            [generate_user_report.s(uid, year, month) for uid in batch],
            aggregate_monthly_reports.s()
        ).apply_async()

9.3 案例三：视频转码服务

"""
视频转码服务

需求：用户上传视频后，转码为多种分辨率（1080p/720p/480p）

策略：
1. 上传后触发转码任务
2. 三个分辨率并行转码
3. 全部完成后发送通知
"""
import subprocess
from celery import chord, group


@app.task(queue='high_priority')
def extract_metadata(video_path: str) -> dict:
    """提取视频元数据"""
    result = subprocess.run(
        ['ffprobe', '-v', 'quiet', '-print_format', 'json', '-show_format', '-show_streams', video_path],
        capture_output=True, text=True
    )
    import json
    metadata = json.loads(result.stdout)
    return {
        'path': video_path,
        'duration': float(metadata['format']['duration']),
        'bitrate': int(metadata['format']['bit_rate']),
    }


@app.task(queue='default')
def transcode_video(video_path: str, resolution: str, video_id: int) -> dict:
    """
    转码视频到指定分辨率

    参数:
        video_path: 源视频路径
        resolution: 目标分辨率（1080p/720p/480p）
        video_id: 视频 ID
    """
    output_path = f'/var/videos/{video_id}_{resolution}.mp4'
    # FFmpeg 转码
    subprocess.run([
        'ffmpeg', '-i', video_path,
        '-s', resolution,
        '-c:v', 'libx264',
        '-crf', '23',
        '-c:a', 'aac',
        '-strict', 'experimental',
        output_path,
        '-y'
    ], check=True)

    # 获取转码后文件大小
    import os
    size = os.path.getsize(output_path)
    return {'video_id': video_id, 'resolution': resolution, 'path': output_path, 'size': size}


@app.task(queue='email')
def notify_video_ready(results: list, video_id: int, user_id: int):
    """所有分辨率转码完成后通知用户"""
    user = User.objects.get(id=user_id)
    resolutions = [r['resolution'] for r in results]
    send_mail(
        subject=f'视频转码完成 #{video_id}',
        message=f'您的视频已转码为：{", ".join(resolutions)}',
        from_email='video@example.com',
        recipient_list=[user.email],
    )
    # 更新数据库状态
    Video.objects.filter(id=video_id).update(status='ready')


def process_video_upload(video_path: str, user_id: int):
    """视频上传后触发转码"""
    video = Video.objects.create(path=video_path, user_id=user_id, status='processing')
    video_id = video.id

    # 工作流：提取元数据 → 并行转码三种分辨率 → 通知用户
    workflow = chain(
        extract_metadata.s(video_path),
        # 三个分辨率并行转码
        chord(
            group(
                transcode_video.s(video_path, '1920x1080', video_id),
                transcode_video.s(video_path, '1280x720', video_id),
                transcode_video.s(video_path, '854x480', video_id),
            ),
            notify_video_ready.s(video_id, user_id)
        )
    )
    workflow.apply_async()
    return video_id

9.4 案例四：分布式爬虫

"""
分布式爬虫系统

需求：爬取 100 个网站的数据

策略：
1. 使用 group 分配任务给多个 Worker
2. 使用 chord 聚合结果
3. 失败任务自动重试
"""
import requests
from bs4 import BeautifulSoup
from celery import group, chord


@app.task(bind=True, max_retries=3, queue='default')
def crawl_page(self, url: str) -> dict:
    """爬取单个页面"""
    try:
        response = requests.get(url, timeout=30)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, 'html.parser')
        return {
            'url': url,
            'title': soup.title.string if soup.title else '',
            'links': [a.get('href') for a in soup.find_all('a') if a.get('href')],
            'status': response.status_code,
        }
    except requests.RequestException as exc:
        raise self.retry(exc=exc, countdown=60)


@app.task(queue='default')
def save_crawl_results(results: list) -> dict:
    """保存爬取结果"""
    from myapp.models import CrawlResult
    for result in results:
        CrawlResult.objects.create(**result)
    return {'total': len(results)}


def crawl_websites(urls: list):
    """分布式爬取多个网站"""
    # 并行爬取所有 URL
    workflow = chord(
        group(crawl_page.s(url) for url in urls),
        save_crawl_results.s()
    )
    result = workflow.apply_async()
    return result.id


# 动态扩展：发现新链接后继续爬取
@app.task(queue='default')
def crawl_with_depth(url: str, depth: int = 1, max_depth: int = 2):
    """带深度的爬取"""
    if depth > max_depth:
        return

    result = crawl_page.delay(url).get()
    save_crawl_results.delay([result])

    if depth < max_depth:
        # 对新发现的链接继续爬取
        for link in result['links'][:10]:  # 限制每页最多爬 10 个链接
            crawl_with_depth.delay(link, depth + 1, max_depth)

10. 习题

10.1 基础题

题目 1：解释 delay() 与 apply_async() 的区别，并给出使用场景。

参考答案要点：

• delay() 是 apply_async() 的语法糖，不支持额外参数
• apply_async() 可指定 countdown、eta、queue、priority 等
• 简单调用用 delay()，需要配置时用 apply_async()

题目 2：列出 Celery 任务的七种状态及其转换关系。

参考答案要点：

• PENDING：任务已发送但未开始
• STARTED：Worker 已开始执行（需 task_track_started=True）
• SUCCESS：任务成功完成
• FAILURE：任务执行失败
• RETRY：任务正在重试
• REVOKED：任务被撤销
• REJECTED：任务被 Worker 拒绝
• 转换关系：PENDING → STARTED → SUCCESS/FAILURE/RETRY，任意状态 → REVOKED

题目 3：编写一个发送邮件的任务，要求失败时自动重试 3 次，每次间隔 60 秒。

参考答案要点：

• 使用 @app.task(bind=True, max_retries=3)
• except 块中调用 self.retry(exc=exc, countdown=60)
• 或使用 autoretry_for=(Exception,), retry_backoff=False, default_retry_delay=60

10.2 进阶题

题目 4：使用 chain 实现以下工作流：下载数据 → 解析数据 → 存储到数据库 → 发送通知。

参考答案要点：

workflow = chain(
    download_data.s(url),
    parse_data.s(),
    store_data.s(),
    send_notification.s()
)
result = workflow.apply_async()

题目 5：解释 worker_prefetch_multiplier 的作用，分析设为 1 与设为 4 的差异。

参考答案要点：

• worker_prefetch_multiplier 控制每个 Worker 子进程预取的消息数
• 设为 1：每次只取一条任务，处理完才取下一条，公平调度，但吞吐量低
• 设为 4（默认）：一次取 4 条任务，吞吐量高，但可能导致长任务阻塞短任务
• 长任务场景建议设为 1

题目 6：为什么 Celery 任务需要幂等性？给出三种实现幂等性的方法。

参考答案要点：

• 原因：Celery 基于”至少一次”投递语义，任务可能被重复执行
• 方法1：唯一 ID 检查（任务开始前查询是否已处理）
• 方法2：数据库唯一约束（插入重复记录时捕获 IntegrityError）
• 方法3：状态机（业务对象仅在特定状态下执行操作）

10.3 挑战题

题目 7：设计一个支持优先级与延迟任务的订单处理系统，要求：

• 紧急订单优先处理
• 普通订单延迟 5 分钟处理
• 失败订单自动重试，指数退避

参考答案要点：

• 配置多队列：high_priority、normal、retry
• 紧急订单：apply_async(queue='high_priority')
• 普通订单：apply_async(countdown=300)
• 重试：self.retry(exc=exc, countdown=5 * (2 ** self.request.retries))

题目 8：分析以下代码的潜在问题，并给出修复方案：

@app.task
def process_order(order: Order):
    payment = charge_payment(order)
    order.status = 'paid'
    order.save()
    send_email(order.user.email)

参考答案要点：

• 问题1：传递 ORM 对象无法序列化
• 问题2：无幂等性，重试时重复扣款
• 问题3：无异常处理，失败后无法重试
• 修复：传递 order_id；幂等性检查；try/except + self.retry()

题目 9：对比 Celery 与 Kafka Streams 在流处理场景下的优劣，并给出选型建议。

参考答案要点：

• Celery：任务粒度大，适合离散任务；有 Result Backend；支持复杂工作流
• Kafka Streams：流式处理，低延迟；无 Result Backend；支持窗口、聚合
• Celery 适合：任务队列、定时调度、批处理
• Kafka Streams 适合：实时流处理、事件溯源、复杂事件处理（CEP）

11. 参考文献

[1] Ask Solem Hoel. 2009. Celery: Distributed Task Queue. Retrieved July 21, 2026, from https://docs.celeryq.dev/

[2] Celery Project. 2024. Celery Documentation v5.4. Retrieved July 21, 2026, from https://docs.celeryq.dev/en/stable/

[3] VMware Inc. 2024. RabbitMQ Documentation: AMQP 0-9-1 Model. Retrieved July 21, 2026, from https://www.rabbitmq.com/tutorials/amqp-concepts.html

[4] Redis Ltd. 2024. Redis Documentation: Pub/Sub and Lists. Retrieved July 21, 2026, from https://redis.io/docs/manual/pubsub/

[5] OASIS. 2014. Advanced Message Queuing Protocol (AMQP) Version 1.0. DOI: 10.1155/2014/354629. Retrieved July 21, 2026, from https://docs.oasis-open.org/amqp/core/v1.0/amqp-core-complete-v1.0.pdf

[6] Brendan Burns and Theodoros Rekatsinas. 2014. At-least-once delivery in distributed message queues. In Proceedings of the 2014 ACM SIGCOMM conference. DOI: 10.1145/2619239.2626293

[7] Ask Solem Hoel. 2020. Celery 5.0 Release Notes. Retrieved July 21, 2026, from https://docs.celeryq.dev/en/stable/history/whatsnew-5.0.html

[8] Rémi Rampin. 2018. Flower: Real-time monitor and web admin for Celery. Retrieved July 21, 2026, from https://flower.readthedocs.io/

[9] Salvatore Sanfilippo. 2024. Redis Documentation: Distributed Locks with Redis. Retrieved July 21, 2026, from https://redis.io/docs/manual/patterns/distributed-locks/

[10] Python Software Foundation. 2024. multiprocessing — Process-based parallelism. Retrieved July 21, 2026, from https://docs.python.org/3/library/multiprocessing.html

[11] Twitter Inc. 2010. Finagle: A Fault-Tolerant RPC System. Retrieved July 21, 2026, from https://finagle.github.io/

[12] Apache Software Foundation. 2024. Apache Kafka Documentation: Streams API. Retrieved July 21, 2026, from https://kafka.apache.org/documentation/streams/

[13] Martin Kleppmann. 2017. Designing Data-Intensive Applications. O’Reilly Media, Sebastopol, CA, USA. Chapter 11: Stream Processing. DOI: 10.5555/3156360

[14] Shiju Sreedharan. 2018. Celery: Distributed Task Queue (3rd ed.). O’Reilly Media. DOI: 10.5555/3273630

[15] Patrick Deziel. 2024. Dramatiq: Fast and reliable distributed task processing library. Retrieved July 21, 2026, from https://dramatiq.io/

12. 延伸阅读

12.1 官方文档

• Celery 官方文档: https://docs.celeryq.dev/
• Celery GitHub 仓库: https://github.com/celery/celery
• Kombu（Celery 底层消息库）: https://docs.celeryq.dev/projects/kombu/en/latest/
• Flower 监控工具: https://flower.readthedocs.io/

12.2 经典教材

• Martin Kleppmann. Designing Data-Intensive Applications, Chapter 11: Stream Processing
• Shiju Sreedharan. Celery: Distributed Task Queue (3rd ed.)
• Salvatore Sanfilippo. Redis in Action

12.3 框架源码

• Celery 源码结构: celery/app/, celery/worker/, celery/canvas.py
• Kombu 消息抽象: kombu/messaging.py, kombu/queues.py
• Beat 调度器: celery/beat.py
• Worker 并发池: celery/concurrency/

12.4 前沿论文与讨论

• At-least-once delivery semantics in distributed systems
• Exponential backoff and jitter in distributed systems (AWS Architecture Blog)
• Celery 6.x 路线图与 asyncio 支持

12.5 相关主题

• python/Python与Redis: Redis 作为 Broker 与 Backend
• python/Python与消息队列: AMQP 协议与消息中间件
• python/Python与Docker: 容器化部署 Celery
• python/Python与Django: Django 项目中集成 Celery