#FastAPIasync-task-queue-celery完全指南

# celery_app.py
from celery import Celery
import time

# 创建 Celery 实例，broker 和 backend 可以分别指定不同的 Redis 数据库
celery = Celery(
    'daoman_fastapi_celery',
    broker='redis://localhost:6379/0',   # 任务队列存储在 Redis 0 号库
    backend='redis://localhost:6379/1'   # 结果存放在 Redis 1 号库（隔离，便于管理）
)

# 定义一个模拟邮件发送任务
@celery.task(bind=True, name="tasks.send_email")
def send_email(self, to: str, subject: str, body: str):
    """
    异步发送邮件，附带进度上报与自动重试
    """
    try:
        # 更新任务状态为 PROGRESS，并告知当前步骤
        self.update_state(state='PROGRESS', meta={'step': 'preparing'})
        time.sleep(1)   # 模拟准备阶段耗时

        self.update_state(state='PROGRESS', meta={'step': 'sending'})
        time.sleep(2)   # 模拟实际发送

        # 成功完成，返回结果字典
        return {'status': 'success', 'to': to, 'time': time.time()}
    except Exception as exc:
        # 失败时自动重试：延迟60秒，最多重试3次
        raise self.retry(exc=exc, countdown=60, max_retries=3)

# main.py
from fastapi import FastAPI, HTTPException
from celery.result import AsyncResult
from celery_app import celery, send_email
from pydantic import BaseModel, EmailStr

app = FastAPI(title="道满FastAPI-Celery教程")

# 请求体模型，利用 Pydantic 自动校验入参
class EmailReq(BaseModel):
    to: EmailStr
    subject: str
    body: str

@app.post("/tasks/email")
async def create_email_task(req: EmailReq):
    # 将任务推入队列，传入收件人、主题、正文
    task = send_email.delay(req.to, req.subject, req.body)
    return {
        "task_id": task.id,
        "status": "submitted",
        "message": "邮件已进入队列，预计3秒左右完成"
    }

@app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    task = AsyncResult(task_id, app=celery)
    # 根据任务状态返回不同的信息
    return {
        "task_id": task_id,
        "state": task.state,
        "info": task.info if task.state in ['PROGRESS', 'FAILURE'] else None,
        "result": task.result if task.state == 'SUCCESS' else None
    }

# config/celery_config.py
from kombu import Queue
import os

class CeleryConfig:
    # ---------- Broker & Backend ----------
    BROKER_URL = os.getenv('CELERY_BROKER', 'redis://localhost:6379/0')
    RESULT_BACKEND = os.getenv('CELERY_BACKEND', 'redis://localhost:6379/1')
    
    # ---------- 序列化方式 ----------
    TASK_SERIALIZER = 'json'
    ACCEPT_CONTENT = ['json']
    RESULT_SERIALIZER = 'json'
    
    # ---------- 时区 ----------
    TIMEZONE = 'Asia/Shanghai'
    ENABLE_UTC = False
    
    # ---------- 任务执行控制 ----------
    TASK_TRACK_STARTED = True          # 开启 STARTED 状态跟踪
    TASK_TIME_LIMIT = 300               # 硬超时5分钟（强制杀死子进程）
    TASK_SOFT_TIME_LIMIT = 240          # 软超时4分钟（会抛出 SoftTimeLimitExceeded 异常，可在任务内处理）
    
    # ---------- 重试策略 ----------
    TASK_RETRY_BACKOFF = True           # 指数退避：重试间隔逐渐增加，防止雪崩
    TASK_RETRY_BACKOFF_MAX = 700        # 最大退避间隔700秒
    TASK_RETRY_JITTER = True            # 随机抖动：避免大量任务在同一时刻集中重试
    
    # ---------- Worker 优化 ----------
    WORKER_PREFETCH_MULTIPLIER = 1      # 每个 Worker 一次只预取1个任务，防止任务积压在某个 Worker 上
    WORKER_MAX_TASKS_PER_CHILD = 1000   # 子进程处理1000个任务后重启，避免内存泄漏
    WORKER_DISABLE_RATE_LIMITS = True   # 在应用层不限制速率（利用 Redis 更快）
    
    # ---------- 队列路由：按任务类型分流 ----------
    TASK_QUEUES = (
        Queue('default', routing_key='default'),
        Queue('ai_tasks', routing_key='ai.#'),
        Queue('email_tasks', routing_key='email.#'),
    )
    TASK_ROUTES = {
        'tasks.ai.*': {'queue': 'ai_tasks'},      # AI相关任务进入 ai_tasks 队列
        'tasks.email.*': {'queue': 'email_tasks'}, # 邮件任务进入 email_tasks 队列
    }

from config.celery_config import CeleryConfig

celery = Celery('daoman_fastapi_celery')
celery.config_from_object(CeleryConfig)

# celery_beat.py
from celery.schedules import crontab
from celery_app import celery
import os

celery.conf.beat_schedule = {
    # 每天凌晨3点清理过期临时文件
    "cleanup-temp-files": {
        "task": "tasks.maintenance.cleanup_temp",
        "schedule": crontab(hour=3, minute=0),
        "options": {"queue": "maintenance"}
    },
    # 每15分钟检查队列健康状态
    "check-queue-health": {
        "task": "tasks.monitoring.check_queue",
        "schedule": crontab(minute="*/15"),
        "options": {"queue": "monitoring"}
    }
}