如何拆分FastAPI服务与Celery Worker服务的目录结构实现独立部署

你要实现API和Worker的无依赖拆分，核心是把Celery的公共配置和任务标识抽成独立的公共包，两边通过任务名称匹配调用，完全不需要互相依赖业务代码，具体方案如下：

API侧不需要知道任务的具体实现，只需要通过统一的Celery实例和任务名称向Broker发任务指令；Worker侧不需要加载FastAPI相关代码，只需要注册对应名称的任务实现，从Broker拉取任务执行，两边通过公共包保证配置和任务名称的一致性。

三个模块完全独立，可单独打包、单独部署，互不依赖：

project-root/
├── fastapi-server/  # 仅部署在API节点，无任何任务执行依赖
│   ├── app/
│   │   ├── models/
│   │   │   ├── __init__.py
│   │   │   ├── model1.py
│   │   │   └── model2.py
│   │   ├── routers/
│   │   │   ├── __init__.py
│   │   │   ├── router1.py
│   │   │   └── router2.py
│   │   └── main.py
│   ├── requirements.txt  # 仅含FastAPI、Uvicorn、Celery核心依赖
│   └── .env  # 统一的Broker连接配置
├── celery-common/  # 轻量公共包，API和Worker节点都要安装
│   ├── celery_app/
│   │   ├── __init__.py
│   │   └── config.py  # 统一Celery实例、任务名称常量定义
│   └── setup.py  # 打包配置，可直接通过pip安装
└── celery-worker/  # 仅部署在计算节点，无FastAPI相关依赖
    ├── tasks/
    │   ├── __init__.py
    │   └── task_impl.py  # 任务的完整业务实现
    ├── worker.py  # Worker启动入口
    └── requirements.txt  # 含任务执行需要的所有计算依赖

celery-common公共包 #

这是唯一的公共依赖，体积非常轻，不含任何业务逻辑：
celery_app/config.py代码示例：

from celery import Celery
import os

# 全局唯一的Celery实例，API和Worker都用这个实例的配置
celery_app = Celery(
    "business_tasks",
    broker=os.getenv("CELERY_BROKER_URL"),
    backend=os.getenv("CELERY_RESULT_BACKEND", None)
)

# 预定义所有任务的名称常量，两边统一调用避免拼写错误
TASK_TEST_TASK = "tasks.test_task"
# 后续新增任务仅需在此处添加对应名称常量即可

通过setup.py将这个包打包为可安装的Python包，命名比如my-project-celery-common，后续两边都通过pip安装即可。

fastapi-server模块 #

不需要安装任何任务执行需要的计算库，仅需安装公共包和FastAPI相关依赖：
路由中调用任务的代码示例：

from fastapi import APIRouter
from celery_app.config import celery_app, TASK_TEST_TASK

router = APIRouter()

@router.post("/trigger-task")
def trigger_task(param1: str, param2: int):
    # 直接通过send_task发任务，不需要导入任务实现
    task_id = celery_app.send_task(
        name=TASK_TEST_TASK,
        args=(param1,),
        kwargs={"param2": param2}
    )
    return {"task_id": str(task_id)}

API节点启动命令不变：
uvicorn app.main:app --port <PORT> --host <HOST>

celery-worker模块 #

不需要安装FastAPI、Uvicorn相关依赖，仅需安装公共包和任务执行需要的计算库：
tasks/task_impl.py代码示例：

from celery_app.config import celery_app, TASK_TEST_TASK

# 任务名称和公共包中定义的常量完全一致即可
@celery.task(name=TASK_TEST_TASK)
def test_task(param1: str, param2: int):
    # 此处编写耗时的业务逻辑，比如模型推理、大数据处理等
    # 完全不需要导入任何FastAPI相关代码
    return "task finished"

worker.py代码示例：

from celery_app.config import celery_app
# 导入所有任务实现，让Celery发现任务
import tasks.task_impl

if __name__ == "__main__":
    celery_app.start()

Worker节点启动命令：
celery -A worker.celery_app worker --loglevel=info

• 参数校验前置：API侧接收请求后先完成所有参数合法性校验，再发送到Broker，避免Worker节点浪费算力处理非法请求
• 序列化配置：统一使用json或msgpack作为任务序列化方式，不要用pickle，避免安全隐患和版本兼容问题
• 队列隔离：不同优先级、不同类型的任务分配到不同队列，Worker节点可指定仅消费特定队列，方便资源调度
• 版本管理：celery-common包做语义化版本管理，升级时两边同步更新，避免任务名称不匹配的问题
• 监控告警：接入Flower监控Celery集群运行状态，配置任务超时、失败告警规则

内容的提问来源于stack exchange，提问作者Shornabho