配置开发环境 Celery 在 FastAPI 的 lifespan 启动时自动开启结束时自动关闭生产环境中 Celery Worker 的独立进程启动方式在 FastAPI 的lifespan中直接启动 Celery Worker 仅适用于开发环境或单进程演示场景。生产环境中Celery Worker 必须作为独立进程运行否则会导致资源竞争、信号处理异常及性能瓶颈。创建celeryconfig.py文件import logging import threading from celery import Celery from celery.worker import state as worker_state from config.configure import ( CELERY_CONCURRENCY, REDIS_DB, REDIS_HOST, REDIS_PORT, ) logger logging.getLogger(__name__) REDIS_URL fredis://{REDIS_HOST}:{REDIS_PORT} celery_app Celery( python_mysql, brokerf{REDIS_URL}/{REDIS_DB}, backendf{REDIS_URL}/{REDIS_DB 1}, # 任务结果存相邻 DB避免与业务缓存混库 include[__name__], # worker 启动时自动导入本模块注册任务 # __name__ 同时兼容 celery -A celeryconfig # 与 config.celeryconfig 两种导入方式 ) celery_app.conf.update( timezoneAsia/Shanghai, enable_utcFalse, task_track_startedTrue, task_acks_lateTrue, # 任务执行成功后才确认worker 崩溃时任务重新投递 worker_prefetch_multiplier1, # 配合 acks_late避免任务预取堆积 broker_connection_retry_on_startupTrue, **({worker_concurrency: CELERY_CONCURRENCY} if CELERY_CONCURRENCY else {}), ) celery_app.task( autoretry_for(Exception,), # 抛出指定异常时自动重试 retry_backoff5, # 指数退避第 n 次重试等待 5 * 2^(n-1) 秒 retry_backoff_max60, # 退避上限 60 秒 retry_jitterTrue, # 加入随机抖动避免重试雪崩 retry_kwargs{max_retries: 3}, # 最多重试 3 次 ) def example_task(x: int, y: int) - int: 示例任务两数相加。除 0 触发异常可观察自动重试行为 if x 0 and y 0: raise ValueError(x 和 y 不能同时为 0) # 用于演示自动重试 result x y logger.info(example_task: %s %s %s, x, y, result) return result # ---------------------------------------------------------------------------- # 内嵌 worker随 FastAPI lifespan 自动启动/关闭无需单独执行 celery 命令 # ---------------------------------------------------------------------------- # macOS 上 billiard 默认 spawn 子进程内嵌时用 solo 池同进程线程内执行最稳妥 EMBEDDED_WORKER_POOL solo EMBEDDED_WORKER_CONCURRENCY 1 _embedded_worker None _worker_thread None _worker_ready threading.Event() def _run_embedded_worker() - None: 在独立守护线程中构造并运行 worker阻塞至 worker 退出。 global _embedded_worker try: worker celery_app.Worker( loglevelinfo, poolEMBEDDED_WORKER_POOL, concurrencyEMBEDDED_WORKER_CONCURRENCY, ) _embedded_worker worker _worker_ready.set() # 通知主线程worker 已构造完成 worker.start() # 阻塞运行stop() 后返回 except Exception: logger.exception(内嵌 Celery worker 运行异常) _worker_ready.set() def start_embedded_worker(timeout: float 10.0) - None: FastAPI lifespan 启动时调用在守护线程中拉起 Celery worker。 global _worker_thread if _worker_thread is not None and _worker_thread.is_alive(): logger.warning(Celery worker 已在运行跳过重复启动) return worker_state.should_stop None # 复位关闭标志防止历史状态残留 _worker_ready.clear() _worker_thread threading.Thread( target_run_embedded_worker, namecelery-worker, daemonTrue, # 守护线程异常情况下也不会拖住进程退出 ) _worker_thread.start() _worker_ready.wait(timeouttimeout) if _embedded_worker is None: logger.error(Celery 内嵌 worker 启动失败请检查上方异常日志) else: logger.info(Celery 内嵌 worker 已启动pool%s, EMBEDDED_WORKER_POOL) def stop_embedded_worker(timeout: float 10.0) - None: FastAPI lifespan 关闭时调用优雅停止 worker 并等待线程退出。 global _embedded_worker, _worker_thread worker, thread _embedded_worker, _worker_thread if worker is None or thread is None: return # 置位 should_stop事件循环与 broker 重连循环会据此退出 # 即便 worker 尚处在启动连接 broker阶段也能被打断 worker_state.should_stop True try: worker.stop() # 优雅关闭warm shutdown except Exception: logger.exception(优雅关闭 Celery worker 失败尝试强制终止) worker.terminate() thread.join(timeouttimeout) if thread.is_alive(): logger.warning(Celery worker 未在 %ss 内退出守护线程将随进程结束, timeout) _embedded_worker None _worker_thread None logger.info(Celery 内嵌 worker 已关闭) 方式一推荐已接入 main.py 的 lifespan 随 FastAPI 启动自动开启、关闭自动结束无需手动管理进程。 方式二独立进程项目根目录执行 celery -A config.celeryconfig worker -l info -P solo 注意macOS 上 billiard 默认 spawn 子进程prefork 池无法继承任务注册表 必须用 -P solo单进程本地验证或 -P threads多线程支持并发。 调用任务Python 侧 from config.celeryconfig import example_task result example_task.delay(1, 2) # 异步投递 print(result.get(timeout10)) # 阻塞获取结果 在main.py里from contextlib import asynccontextmanager from config.database import init_db, close_db from config.redis import close_redis asynccontextmanager async def life_span(app: FastAPI): print(server is starting …) await init_db() # 仅开发模式CELERY_EMBEDDEDtrue默认内嵌 worker # 生产环境使用独立进程scripts/celery_worker.sh if CELERY_EMBEDDED: start_embedded_worker() else: logging.info( CELERY_EMBEDDEDfalse跳过内嵌 worker请确认独立 Celery 进程已启动 ) yield # 先停 Celery需使用 Redis broker再关闭 Redis/DB 连接 if CELERY_EMBEDDED: stop_embedded_worker() await close_db() await close_redis() print(server has been shut down.) app FastAPI(lifespanlife_span)在根目录创建脚本文件celery_worker.sh和.envcelery_worker.sh#!/usr/bin/env bash # # Celery worker 独立进程管理脚本生产部署 / macOS 本地验证均可用 # # 用法项目根目录执行 # scripts/celery_worker.sh start # 后台启动 worker单实例重复启动会被拒绝 # scripts/celery_worker.sh stop # 优雅停止SIGTERM warm shutdown超时再 KILL # scripts/celery_worker.sh restart # 等同于 stop start # scripts/celery_worker.sh status # 查看运行状态退出码 0运行中3未运行 # # 可选环境变量也可写在 .env 中由 config.configure 读取CLI 参数优先级更高 # CELERY_POOLsolo|prefork|threads # 不设置时Linuxprefork(默认)macOSsolo # CELERY_CONCURRENCY4 # worker 并发数默认按 CPU 核数 # CELERY_QUEUEScelery # 监听队列多个用逗号分隔 # CELERY_HOSTNAMEworker%h-1 # 节点名多实例部署时区分 # CELERY_LOG_LEVELinfo # 日志级别 # # 退出码0 成功1 参数或启动失败3 未在运行(status) # set -euo pipefail APPconfig.celeryconfig # 定位项目根目录脚本位于 root/scripts/ 下 PROJECT_ROOT$(cd $(dirname ${BASH_SOURCE[0]})/.. pwd) cd $PROJECT_ROOT # 优先使用项目虚拟环境内的 celery其次使用 PATH 中的 CELERY_BIN$PROJECT_ROOT/venv/bin/celery if [ ! -x $CELERY_BIN ]; then CELERY_BIN$(command -v celery) fi if [ -z $CELERY_BIN ]; then echo 未找到 celery 可执行文件请先安装依赖或激活虚拟环境 2 exit 1 fi # 运行参数 -------------------------------------------------------------- POOL${CELERY_POOL:-} # macOS 上 billiard 默认 spawnprefork 无法继承任务注册表自动回退 solo if [ -z $POOL ] [ $(uname -s) Darwin ]; then POOLsolo fi CONCURRENCY${CELERY_CONCURRENCY:-} QUEUES${CELERY_QUEUES:-celery} HOSTNAME${CELERY_HOSTNAME:-} LOG_LEVEL${CELERY_LOG_LEVEL:-info} RUN_DIR$PROJECT_ROOT/run LOG_DIR$PROJECT_ROOT/logs PIDFILE$RUN_DIR/celery_worker.pid LOGFILE$LOG_DIR/celery_worker.log # 停止时最长等待秒数warm shutdownprefork 下会等在途任务执行完 STOP_TIMEOUT30 mkdir -p $RUN_DIR $LOG_DIR is_running() { [ -f $PIDFILE ] || return 1 local pid pid$(cat $PIDFILE 2/dev/null || true) [ -n $pid ] kill -0 $pid 2/dev/null } build_args() { ARGS(-A $APP worker -l $LOG_LEVEL -Q $QUEUES --pidfile $PIDFILE --logfile $LOGFILE) if [ -n $POOL ]; then ARGS(-P $POOL) fi if [ -n $CONCURRENCY ]; then ARGS(-c $CONCURRENCY) fi if [ -n $HOSTNAME ]; then ARGS(-n $HOSTNAME) fi } start_worker() { if is_running; then echo celery worker 已在运行 (pid$(cat $PIDFILE))跳过启动 return 0 fi # 清理上次异常退出残留的失效 pidfile rm -f $PIDFILE build_args echo 启动 celery worker: $CELERY_BIN ${ARGS[*]} # nohup 后台运行stdout/stderr 追加到同一日志文件承接 celery logfile 之外的早期输出 nohup $CELERY_BIN ${ARGS[]} $LOGFILE 21 # 等待 celery 写入 pidfile 且进程存活约 10s for _ in $(seq 1 50); do if is_running; then echo celery worker 已启动 (pid$(cat $PIDFILE))日志: $LOGFILE return 0 fi sleep 0.2 done echo celery worker 启动失败请查看日志: $LOGFILE 2 exit 1 } stop_worker() { if ! is_running; then echo celery worker 未在运行 rm -f $PIDFILE return 0 fi local pid pid$(cat $PIDFILE) echo 优雅停止 celery worker (pid$pid)最长等待 ${STOP_TIMEOUT}s ... # SIGTERMCelery warm shutdown等待在途任务执行完毕后退出 kill -TERM $pid 2/dev/null || true local waited0 while kill -0 $pid 2/dev/null; do if [ $waited -ge $STOP_TIMEOUT ]; then echo 等待超时强制结束 worker 进程组 ... 2 # 兜底仅精确匹配本项目虚拟环境 本 app 的 worker 进程避免误杀其他项目 pkill -KILL -f $CELERY_BIN.*-A $APP worker 2/dev/null || true break fi sleep 1 waited$((waited 1)) done rm -f $PIDFILE echo celery worker 已停止 } status_worker() { if is_running; then echo celery worker 运行中 (pid$(cat $PIDFILE)) exit 0 fi echo celery worker 未运行 exit 3 } case ${1:-} in start) start_worker ;; stop) stop_worker ;; restart) stop_worker start_worker ;; status) status_worker ;; *) echo 用法: $0 {start|stop|restart|status} 2 exit 1 ;; esac.envREDIS_HOSTlocalhost REDIS_PORT6379 # Celery 运行模式true内嵌 worker本地开发随 FastAPI 自动启停 # 生产环境改为 falseworker 用 scripts/celery_worker.sh 独立进程管理 CELERY_EMBEDDEDtrue