# Celery 异步任务规范

> **版本**: v1.0 | **修订日期**: 2026-07-15
> **技术栈**: Celery 5.4+ / Redis (Broker + Backend)

---

## 一、Celery 实例配置

```python
# app/tasks/celery_app.py
from celery import Celery
from app.core.config import settings

celery_app = Celery(
    "worker",
    broker=settings.CELERY_BROKER_URL,
    backend=settings.CELERY_RESULT_BACKEND,
)

celery_app.conf.update(
    # 序列化
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",

    # 时区
    timezone="Asia/Shanghai",
    enable_utc=True,

    # 任务执行
    task_track_started=True,
    task_time_limit=300,          # 硬超时 5 分钟
    task_soft_time_limit=240,     # 软超时 4 分钟

    # Worker（SQLite 场景）
    worker_concurrency=1,         # SQLite 不支持并发写
    worker_prefetch_multiplier=1,
    worker_pool="solo",           # 单进程池

    # 结果过期
    result_expires=3600,

    # Beat 定时任务
    beat_schedule={},
)
```

---

## 二、任务定义规范

### 2.1 命名规则

```
{action}_{entity}_task
```

示例: `sync_inventory_task`, `generate_report_task`, `send_notification_task`

### 2.2 任务定义

```python
# app/tasks/inventory_tasks.py
from app.tasks.celery_app import celery_app

@celery_app.task(
    name="inventory.sync_inventory",
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError),
    max_retries=3,
    retry_backoff=True,
    retry_backoff_max=600,
)
def sync_inventory_task(self, warehouse_id: int, material_type: str | None = None):
    """同步库存数据"""
    try:
        # 任务逻辑（Celery 任务内使用同步数据库会话）
        from app.core.database import SessionLocal
        with SessionLocal() as db:
            # ... 业务逻辑
            db.commit()
    except Exception as exc:
        self.retry(exc=exc)
```

### 2.3 参数规范

| 规则           | 说明                                           |
|----------------|------------------------------------------------|
| 可序列化       | 参数必须是 JSON 可序列化的（str/int/float/dict/list） |
| 禁止传对象     | 禁止传递 SQLAlchemy Model 实例                  |
| 传递 ID        | 通过 ID 在任务内部重新查询数据库                 |
| 参数校验       | 任务开头进行参数校验，失败直接 raise             |

---

## 三、任务调用

```python
from app.tasks.inventory_tasks import sync_inventory_task

# 异步调用
result = sync_inventory_task.delay(warehouse_id=123, material_type="aluminum")

# 指定 ETA
from datetime import datetime, timedelta
result = sync_inventory_task.apply_async(
    kwargs={"warehouse_id": 123},
    eta=datetime.utcnow() + timedelta(minutes=10),
)

# 获取结果（同步等待）
result.get(timeout=60)

# 检查状态
result.status  # PENDING / STARTED / SUCCESS / FAILURE / RETRY
result.ready() # bool
```

---

## 四、错误处理与重试

```python
@celery_app.task(
    bind=True,
    autoretry_for=(ConnectionError,),
    max_retries=3,
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
)
def process_order_task(self, order_id: int):
    """处理订单"""
    try:
        # 业务逻辑
        ...
    except BusinessLogicError as exc:
        # 业务错误不重试，直接记录
        logger.error(f"Order {order_id} failed: {exc}")
        raise
    except ExternalServiceError as exc:
        # 外部服务错误，手动重试
        raise self.retry(exc=exc, countdown=60)
```

| 场景           | 策略                 |
|----------------|----------------------|
| 网络超时       | autoretry + backoff  |
| 业务逻辑错误   | 不重试，记录日志     |
| 数据不一致     | 不重试，告警通知     |
| 外部服务故障   | 手动 retry + 固定间隔|

---

## 五、定时任务（Celery Beat）

```python
celery_app.conf.beat_schedule = {
    "daily-inventory-sync": {
        "task": "inventory.sync_inventory",
        "schedule": crontab(hour=2, minute=0),  # 每天凌晨 2 点
        "kwargs": {"warehouse_id": 0},
    },
    "weekly-report": {
        "task": "report.generate_weekly_report",
        "schedule": crontab(day_of_week="monday", hour=8, minute=0),
    },
}
```

---

## 六、Worker 配置

### 6.1 启动命令

```bash
# 开发环境
celery -A app.tasks.celery_app worker --loglevel=debug --pool=solo

# 生产环境（SQLite）
celery -A app.tasks.celery_app worker --loglevel=info --pool=solo --concurrency=1

# Beat 定时任务调度
celery -A app.tasks.celery_app beat --loglevel=info
```

### 6.2 SQLite 注意事项

| 事项         | 说明                                           |
|--------------|------------------------------------------------|
| 并发         | 必须 `--concurrency=1`，SQLite 不支持并发写    |
| 连接池       | 每个任务创建新的数据库连接，任务结束关闭       |
| 事务         | 任务内部手动 commit，不依赖外部事务            |
| 长时间任务   | 拆分为多个短任务，避免长时间持有数据库锁       |

---

## 七、监控（Flower）

```bash
celery -A app.tasks.celery_app flower --port=5555
# 访问 http://localhost:5555
```

---

## 八、任务测试

```python
# tests/conftest.py
@pytest.fixture(autouse=True)
def celery_eager_mode(monkeypatch):
    """测试时任务同步执行"""
    from app.tasks.celery_app import celery_app
    celery_app.conf.update(
        task_always_eager=True,
        task_eager_propagates=True,
    )

# tests/test_tasks/test_inventory.py
def test_sync_inventory_task_success(db_session):
    """测试库存同步任务成功"""
    warehouse = WarehouseFactory()
    db_session.add(warehouse)
    db_session.commit()

    result = sync_inventory_task.delay(warehouse_id=warehouse.id)
    assert result.successful()

def test_sync_inventory_task_invalid_id():
    """测试无效仓库 ID"""
    result = sync_inventory_task.delay(warehouse_id=99999)
    assert result.failed()
```