一、为什么测试平台需要异步任务队列？

在自研自动化测试平台、资源管理系统或性能调度服务中，用户发起的很多操作都属于长耗时任务：

• 点击“立即执行冒烟回归”，需要调用后端跑 10 分钟自动化用例；
• 周期性同步全网虚拟机或测试环境的主机资源（CPU/内存/磁盘状态）；
• 批量生成并发送包含数十兆附件的测试报告邮件。

如果直接在 Web 接口处理函数中同步执行这些操作，HTTP 请求会在几秒后直接触发超时断连，整个 Web 服务工作进程也会被耗尽。

Celery 结合 Redis/RabbitMQ 是 Python 生态中最成熟的分布式异步任务队列解决方案。

---

二、系统架构与多队列（Queue）分流隔离

在大型调度平台中，最忌讳的是**“所有任务挤在同一个默认队列（celery）里”**：

• 如果有一个耗时 30 分钟的慢速全量同步任务正在执行，用户临时下发的单用例调试任务就会在队列中死等排队数十分钟。

队列隔离设计方案：

┌────────────────────────────────────────────────────────┐
│            测试平台 Web 端 (Producer / 任务生产者)       │
└────────────┬──────────────────────────────┬────────────┘
             │ 下发手动即时任务              │ 下发周期长任务
             ▼                              ▼
    [ Redis 队列: test_manual ]    [ Redis 队列: test_sync ]
             │                              │
             ▼                              ▼
┌─────────────────────────┐    ┌─────────────────────────┐
│ Worker 1: 专用即时通道   │    │ Worker 2: 专用后台通道  │
│ (快速消费，保障秒级响应)  │    │ (允许长时间占用运行)     │
└─────────────────────────┘    └─────────────────────────┘

---

三、Celery 核心配置与任务调度实现

1. 任务定义与路由分流配置

from celery import Celery
from datetime import timedelta

app = Celery("test_platform_tasks", broker="redis://127.0.0.1:6379/1", backend="redis://127.0.0.1:6379/2")

# 配置队列路由规则
app.conf.task_routes = {
    "tasks.execute_manual_case": {"queue": "test_manual"},
    "tasks.sync_all_host_resources": {"queue": "test_sync"},
}

# 配置 Celery Beat 定时周期调度计划
app.conf.beat_schedule = {
    "refresh-host-resources-every-5min": {
        "task": "tasks.sync_all_host_resources",
        "schedule": timedelta(minutes=5), # 每 5 分钟自动执行一次
    },
}

2. 任务执行函数编写

@app.task(bind=True, max_retries=3)
def execute_manual_case(self, case_id: int):
    """手动执行用例任务（走 test_manual 高优先级队列）"""
    print(f"开始执行用例 ID: {case_id}")
    # 模拟自动化测试过程
    return {"case_id": case_id, "status": "PASSED"}

---

四、生产避坑核心实战指南

1. Windows 本地调试深坑：必须指定 --pool=solo

在 Windows 开发机上直接运行 celery worker，默认的 prefork 进程池会因为 Windows 不支持标准的 Unix fork 系统调用而触发底层死锁或无限报错：

# ✅ Windows 下必须显式指定 --pool=solo 单进程运行：
celery -A tasks.app worker -l info --pool=solo --queues=test_manual --concurrency=2

2. 异步任务中数据库 Session 找不到（Web 上下文丢失）

在 Celery 任务内部直接调用 ORM 模型（如 SQLAlchemy db.session）时，经常抛出 RuntimeError: Working outside of application context。

• 解法：在任务执行体内部显式推入 Web 框架上下文：

  from my_web_app import create_app, db
  
  flask_app = create_app()
  
  @app.task
  def sync_all_host_resources():
      # 显式激活 Flask/Django 应用上下文
      with flask_app.app_context():
          # 此时可以安全访问 db.session 和 current_app
          db.session.query(...)

3. 可视化监控运维：集成 Flower

在后台拉起 Flower 容器或守护进程，提供可视化的任务堆积、执行耗时与 Worker 健康状态大盘：

celery -A tasks.app flower --port=5555

访问 http://127.0.0.1:5555，即可实时查看所有队列的吞吐速率、失败任务堆栈并支持一键取消僵死任务。