无法调试subprocess启动的Celery业务代码？求技术解决方案

问题原因 #

你当前的代码通过subprocess.run启动Celery Worker，这会创建一个独立的子进程，而你的调试器只附着在main.py所在的主进程上，子进程未被调试器接管，所以业务代码里的断点不会触发。

方案一：给子进程添加调试附着参数 #

这种方式保留原有的subprocess调用逻辑，通过给Celery命令添加调试启动参数，让调试器能连接到子进程：

1. 先安装调试工具debugpy（Poetry环境下）：

poetry add debugpy --dev

2. 修改封装命令的handle方法里的cmd列表，用当前环境的Python解释器启动debugpy，再加载Celery：

import sys  # 记得导入sys模块

def handle(self, *args, **options):
    service = options.get('service')
    os.environ.setdefault('CELERY_NAME', ','.join(service))
    server_hostname = os.environ.get("SERVER_HOSTNAME")
    if hasattr(os, 'getuid') and os.getuid() == 0:
        os.environ.setdefault('C_FORCE_ROOT', '1')
    if not server_hostname:
        server_hostname = '%h'
    
    # 修改后的cmd，添加debugpy启动参数
    cmd = [
        sys.executable,  # 使用当前Poetry环境的Python解释器
        "-m", "debugpy",
        "--listen", "localhost:5678",  # 监听本地端口
        "--wait-for-client",  # 等待调试客户端连接后再启动Worker
        "-m", "celery",
        '-A', 'ops',
        'worker',
        '-P', 'threads',
        '-l', 'info',
        '-c', '10',
        '-Q', ','.join(service),
        '--heartbeat-interval', '10',
        '-n', f'{",".join(service)}@{server_hostname}',
        '--without-mingle',
    ]
    kwargs = {'cwd': BASE_DIR}

    print("===== args ======")
    print(cmd)
    print("===== args ======")
    print(kwargs)
    subprocess.run(cmd, **kwargs)

3. 配置调试客户端（以PyCharm为例）：

• 打开Run/Debug Configurations，新建Python Remote Debug配置
• 设置Host为localhost，Port为5678
• 先启动main.py的调试，当控制台输出等待客户端连接时，启动刚才创建的Remote Debug配置，此时就能命中业务代码里的断点了。

方案二：直接调用Celery API启动Worker（更简便） #

放弃subprocess调用，直接用Celery的Python API启动Worker，这样Worker和main.py在同一个进程中，调试器能直接捕获断点：

from celery import Celery
import os

def handle(self, *args, **options):
    service = options.get('service')
    os.environ.setdefault('CELERY_NAME', ','.join(service))
    server_hostname = os.environ.get("SERVER_HOSTNAME")
    if hasattr(os, 'getuid') and os.getuid() == 0:
        os.environ.setdefault('C_FORCE_ROOT', '1')
    if not server_hostname:
        server_hostname = '%h'
    
    # 初始化Celery app（根据项目实际配置方式调整）
    app = Celery('ops')
    # 假设项目用Django配置，加载settings里的Celery配置
    app.config_from_object('django.conf:settings', namespace='CELERY')
    # 或者从配置文件加载：app.config_from_file('celeryconfig.py')

    # 直接启动Worker
    worker = app.Worker(
        pool='threads',
        loglevel='info',
        concurrency=10,
        queues=service,
        heartbeat_interval=10,
        hostname=f'{",".join(service)}@{server_hostname}',
        without_mingle=True,
    )
    worker.start()

这样启动后，直接调试main.py就能触发业务代码里的断点，完全符合你的使用场景。

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