2. 异步导出基础：异步任务队列原理

异步导出的核心在于将耗时的模板渲染与文件生成操作从 HTTP 请求-响应周期中剥离，交由后台进程处理。本节将深入剖析异步任务队列的三大支柱：消息代理（RabbitMQ）、任务调度器（Celery）以及任务状态机与回调机制。

2.1 核心组件与架构

一个典型的异步导出系统由以下组件构成：

2.2 消息代理：RabbitMQ 核心原理

RabbitMQ 作为 AMQP（高级消息队列协议）的实现，在异步导出中承担消息路由与缓冲的职责。其核心概念包括：

• Exchange（交换机）：接收生产者消息，根据路由键将消息分发到一个或多个队列。常用类型有 direct、topic、fanout。
• Queue（队列）：存储消息的缓冲区，Worker 从中拉取任务。支持持久化、排他、自动删除等属性。
• Binding（绑定）：定义 Exchange 与 Queue 之间的关联关系，包含路由键匹配规则。
• Message（消息）：包含任务数据（如导出参数、模板 ID）和元数据（如优先级、TTL）。

2.3 任务调度器：Celery 工作机制

Celery 是 Python 生态中最成熟的分布式任务队列框架。其任务生命周期如下：

1. 任务定义：使用 @app.task 装饰器将普通函数注册为 Celery 任务。
2. 任务发布：调用 task.delay() 或 task.apply_async() 将任务消息发送到 Broker。
3. 消息路由：Broker 根据 Exchange 和路由键将消息放入指定队列。
4. 任务执行：Worker 从队列拉取消息，反序列化后执行任务函数。
5. 结果存储：任务完成后，将状态和返回值写入 Result Backend。

# celery_app.py
from celery import Celery

app = Celery('export_tasks',
             broker='amqp://guest:guest@localhost:5672//',
             backend='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def generate_export_report(self, template_id, params):
    try:
        # 执行模板渲染与文件生成
        result = render_and_build(template_id, params)
        return result
    except Exception as exc:
        raise self.retry(exc=exc)

2.4 任务状态机

Celery 为每个任务维护一个明确的状态机，用于追踪任务从创建到完成的完整生命周期。状态转换路径如下：

2.5 回调机制与前端轮询

异步导出完成后，系统需要将结果通知前端。常见的实现方式有两种：

• 前端轮询（Pull）：前端定时调用状态查询接口，检查任务状态。当状态变为 SUCCESS 时，获取下载链接。
• WebSocket 推送（Push）：Worker 完成任务后，通过 WebSocket 或 Server-Sent Events 主动推送结果给前端。

典型的回调流程如下：

1. 前端调用导出接口，后端返回任务 ID（UUID）。
2. 前端启动轮询，每隔 2-5s 调用 GET /api/tasks/{task_id}/status。
3. 后端查询 Result Backend，返回当前状态及进度信息（如 {"status": "PROGRESS", "current": 50, "total": 100}）。
4. 当状态变为 SUCCESS 时，后端返回文件下载 URL，前端触发下载。
5. 若状态变为 FAILURE，前端展示错误提示，并提供重试入口。

# views.py
from celery.result import AsyncResult
from django.http import JsonResponse

def task_status(request, task_id):
    result = AsyncResult(task_id)
    response = {
        'task_id': task_id,
        'status': result.status,
        'result': result.result if result.successful() else None,
    }
    if result.status == 'PROGRESS':
        response.update(result.info)  # 自定义进度信息
    return JsonResponse(response)

2.6 常见问题与避坑指南