Celery 任务测试指南

1. 自动化测试（单元测试/CI）

项目新增了三个演示型任务（api_v1/tasks.py）：

任务	功能
`ping_task(message='ping')`	打印并返回时间戳，用于快速验证 worker 是否可接收/返回结果
`merchant_product_count(merchant_id)`	统计某商户下产品数量，示范如何在任务中访问数据库
`backup_database(output_dir=None, filename_prefix='db-backup')`	将当前数据库导出为 `.sql` 文件（数据备份），默认保存到 `BASE_DIR/data-bak/`

对应的单元测试位于 api_v1/tests.py（类 CeleryTasksTestCase），通过 @override_settings(CELERY_TASK_ALWAYS_EAGER=True) 让任务在测试进程内同步执行。

运行方式：

uv run manage.py test api_v1.tests.CeleryTasksTestCase

若要在其他测试中调用任务，只需在测试类上使用同样的 override_settings 即可保证 Celery 在没有 worker 的情况下仍能同步执行。

2. 本地验证真实 worker（RabbitMQ + Celery）

2.1 启动依赖服务

# 启动基础设施（Postgres/Redis/Rabbit/Celery worker/web）
docker compose up -d postgres redis rabbitmq

# 启动 web（Django）
docker compose up -d web

# 启动 Celery worker（如已运行，可跳过）
docker compose up -d celery_worker
# 或者手动：uv run celery -A flower worker -l info

若 worker 容器异常，可通过 docker compose restart celery_worker 重启。

2.2 发送测试任务

在另一个终端进入容器或宿主项目目录执行：

uv run python manage.py shell

from api_v1.tasks import ping_task, merchant_product_count, backup_database
from basic_info.models import Merchant

# 发送心跳任务
async_result = ping_task.delay('hello celery')
print(async_result.get(timeout=10))

# 发送统计任务（示例：使用 ID=1 的商户）
merchant = Merchant.objects.first()
result = merchant_product_count.delay(merchant.id)
print(result.get(timeout=10))

# 触发数据库备份（输出到默认 data-bak）
backup = backup_database.delay(filename_prefix='manual-backup')
print(backup.get(timeout=30))  # payload 中包含 backup_path，文件为 .sql

2.3 观察执行结果

Worker 日志：docker compose logs -f celery_worker
可查看任务被消费、日志输出等信息。
RabbitMQ 控制台：访问 http://localhost:15672（默认账号 guest/guest），观察队列长度是否回落到 0。
Task Result：上面的 result.get() 会在任务完成时返回 payload，若超时或无法连接则会抛出异常，帮助定位问题。

3. 失败排查 Checklist

环境变量：CELERY_BROKER_URL 和 CELERY_RESULT_BACKEND 是否指向正在运行的 RabbitMQ/Redis？
Worker 进程：docker compose ps 确认 celery_worker 状态为 Up。
队列阻塞：RabbitMQ 控制台查看是否有大量消息处于 Unacked。
日志级别：测试中需要捕获日志时，可使用 with self.assertLogs('api_v1.tasks', level='INFO').
自动化测试：若任务在测试里需要真实队列，请去掉 CELERY_TASK_ALWAYS_EAGER 覆盖；否则保持默认值即可同步执行。
数据库备份依赖：backup_database 在 PostgreSQL 场景下需要系统可执行 pg_dump，请确保容器/宿主机已安装 PostgreSQL 客户端工具；SQLite 则会使用内置 iterdump 生成 .sql。

4. 常用命令速查

操作	命令
启动 worker	`docker compose up -d celery_worker`
查看 worker 日志	`docker compose logs -f celery_worker`
重启 worker	`docker compose restart celery_worker`
清空 RabbitMQ 队列	`docker compose exec rabbitmq rabbitmqctl purge_queue flower`（示例）
运行单个任务测试	`uv run python manage.py shell` -> `ping_task.delay()`

通过以上步骤，即可确认 Celery + RabbitMQ 在本地能够成功执行任务，并在自动化测试里保持覆盖。若需要新增业务任务，可参考 api_v1/tasks.py 的写法：使用 @shared_task，在任务内进行必要的日志记录，便于问题排查。***

4.2 KiB Raw Blame History Unescape Escape