xiaoxia-saas/packages/shared/celery_queues.py

"""Celery 队列定义与路由配置（API / Worker 共享）。

#1714 队列隔离：用户等待的视频生成任务路由到高优先级 `generation` 队列，
由专用 worker 进程独占消费；素材入库/转码等后台批量任务路由到 `transcode`
队列；其余杂项任务走默认 `celery` 队列。转码队列积压时，视频生成任务
仍能被 generation worker 立即领取执行，不会排队。

队列说明：
- generation: 用户提交的视频生成/预览渲染（延迟敏感，资源消耗大）
- transcode: 素材入库（HEVC 转码）、AI 分类、素材查重（批量、可排队）
- celery(默认): 配音、语音、下载缩略图、定时清理等杂项
"""

from __future__ import annotations

from kombu import Queue

# ── 队列名常量（生产端与消费端共用，禁止拼写漂移） ──
QUEUE_GENERATION = "generation"
QUEUE_TRANSCODE = "transcode"
QUEUE_DEFAULT = "celery"

# Worker 消费的队列列表（顺序即优先级：高优队列排在前面）
WORKER_QUEUES = (QUEUE_GENERATION, QUEUE_TRANSCODE, QUEUE_DEFAULT)

# 队列声明：持久化队列，broker 重启不丢消息
task_queues = (
    Queue(QUEUE_GENERATION, routing_key=QUEUE_GENERATION, durable=True),
    Queue(QUEUE_TRANSCODE, routing_key=QUEUE_TRANSCODE, durable=True),
    Queue(QUEUE_DEFAULT, routing_key=QUEUE_DEFAULT, durable=True),
)

# ── 任务路由表：task name → 队列 ──
# 键支持 celery 标准通配符。
task_routes = {
    # 高优先级：用户等待的视频生成
    "worker.generate_video": {"queue": QUEUE_GENERATION},
    # 后台批量：素材入库/转码 + AI 分类 + 素材查重，积压不影响生成
    "worker.ingest_asset": {"queue": QUEUE_TRANSCODE},
    "worker.classify_asset": {"queue": QUEUE_TRANSCODE},
    "worker.process_duplication_check": {"queue": QUEUE_TRANSCODE},
    "worker.check_duplicate": {"queue": QUEUE_TRANSCODE},
}

# 生成任务的预取数：渲染是长任务，预取 1 避免任务被某个 worker 占住不调度
GENERATION_WORKER_PREFETCH_MULTIPLIER = 1


def apply_queue_settings(app) -> None:
    """把队列隔离配置应用到 Celery app（API 生产端与 Worker 消费端都要调用）。

    配置 task_queues / task_routes / task_default_queue。生产端靠 task_routes
    把消息投递到对应队列；消费端靠 task_queues 声明自己消费哪些队列
    （实际消费集由启动参数 -Q 控制）。
    """
    app.conf.task_queues = task_queues
    app.conf.task_routes = task_routes
    app.conf.task_default_queue = QUEUE_DEFAULT