如何通过Python脚本启动Celery worker并在执行完成后将其关闭？

你写的代码无法正常运行核心有三个问题：

• app.worker_main()是Celery命令行入口的Python封装，传入-D（后台分离运行）参数时，方法只会fork子进程启动worker，然后立刻返回，不会等待worker完成初始化、注册到消息Broker。
• 你在同一个进程里紧接着调用shutdown指令时，worker大概率还没启动完成，根本收不到关闭广播；主进程会直接顺序执行完所有代码立刻退出，不仅worker没真正跑起来，中间依赖worker的业务逻辑也会因为worker未就绪执行失败。
• 如果不加-D参数，worker_main()会直接阻塞当前主进程运行worker事件循环，你写在它后面的业务代码要等worker完全退出才会执行，完全不符合预期。

方案1：生产/常规场景推荐（进程隔离+生命周期可控） #

用子进程方式启动独立worker进程，轮询确认worker就绪后执行业务逻辑，最后显式回收进程，避免孤儿进程残留：

import subprocess
import time
from myapp import app  # 导入你项目里的Celery实例
from celery.app.control import Inspect

if __name__ == "__main__":
    # 启动worker进程，不要加-D后台运行参数，方便管理进程
    worker_process = subprocess.Popen(
        ["celery", "-A", "myapp", "worker", "-Q", "myqueue", "--loglevel=info"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL
    )

    # 轮询检测worker是否就绪，最长等待10秒
    max_wait_seconds = 10
    worker_ready = False
    inspector = Inspect(app=app)
    for _ in range(max_wait_seconds):
        try:
            if inspector.ping():  # ping通代表worker已完成启动、可正常接收任务
                worker_ready = True
                break
        except Exception:
            time.sleep(1)
            continue
        time.sleep(1)

    if not worker_ready:
        worker_process.kill()
        raise RuntimeError("Celery worker启动失败，超时未就绪")

    try:
        # 此处编写所有依赖worker运行的业务逻辑
        # 例如任务提交、结果等待等操作
        pass
    finally:
        # 优雅关闭worker，避免进程残留
        active_workers = list(inspector.ping().keys())
        if active_workers:
            app.control.shutdown(destination=active_workers)
        # 等待进程退出，超时强制kill
        try:
            worker_process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            worker_process.kill()
            worker_process.wait()

方案2：本地调试/测试场景用 #

如果只是本地跑测试、写单元测试，可以直接用Celery内置的测试worker，不需要手动管理进程：

if __name__ == "__main__":
    # 启动内存级测试worker，仅适合调试/测试场景，禁止生产使用
    with app.worker_for_testing(queues=["myqueue"]) as test_worker:
        # 块内直接执行业务逻辑即可，退出with块自动关闭worker
        pass

• 不要在同一个Python主进程里直接调用worker_main()启动worker后再写业务逻辑，worker的事件循环会阻塞主进程，后续代码根本没有执行机会。
• 尽量不要用-D后台模式启动内嵌worker，这种模式下你拿不到worker的进程PID，没法精准控制关闭，很容易在脚本退出后残留孤儿worker进程占用资源。

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