如何设置Celery任务超时时间，超时未完成时自动中止任务？

你之前尝试的常规Celery超时方案不生效，基本逃不出两个常见问题：

• 配置未正确加载：仅在全局配置中设置超时参数，容易被任务级参数、worker启动参数覆盖；如果使用gevent/eventlet协程池运行worker，这类池对Celery原生的SIGALRM软超时、SIGKILL硬超时逻辑存在已知兼容问题
• 外部脚本拦截/不响应信号：直接在worker进程内运行的外部脚本如果存在阻塞C扩展调用、死循环不释放GIL、自定义重写信号处理逻辑的情况，Celery发出的超时信号无法被正常响应，自然无法终止任务

按照可靠性从低到高排序，你可以根据实际场景选择：

方案1：修复原生超时配置（最快验证） #

Celery原生支持软、硬两层超时，直接在任务定义处显式声明参数，避免配置覆盖问题，同时加worker启动参数兜底：

1. 改造任务定义，显式指定超时阈值：

from celery.exceptions import SoftTimeLimitExceeded

# 示例配置：软超时300秒触发异常供业务清理，硬超时320秒直接杀进程兜底
@app.task(name="execute_script", bind=True, base=CallbackTask, soft_time_limit=300, time_limit=320)
def execute_script(self, script_id, variables, name, start_time):
    self.update_state(state='EXECUTING')
    script = Script.objects.get(pk=script_id)
    mod = Script.get_script(script)
    try:
        mod.main(variables, name)
    except SoftTimeLimitExceeded:
        # 自定义超时后的清理逻辑：更新任务状态、记录日志、回滚业务数据
        self.update_state(state='FAILURE', meta={'error': '任务执行超时'})
        raise
    except Exception as e:
        raise e
    return name

2. 启动worker时追加超时参数兜底，避免配置读取异常：
   celery -A main worker -l info --soft-time-limit=300 --time-limit=320

注意：该方案仅在使用默认prefork进程池时可靠，如果必须使用协程池，请直接用方案2。

方案2：子进程隔离运行脚本（最推荐，100%可控） #

直接在worker进程内import运行外部脚本风险极高：脚本死循环、内存泄漏、篡改信号处理逻辑都会直接打挂worker。最稳妥的方式是将脚本执行逻辑放到独立子进程中，自主控制超时逻辑，完全不依赖Celery的原生超时机制：

import multiprocessing

def _script_runner(mod, variables, name, result_queue):
    """子进程内实际执行脚本的入口，结果通过队列回传"""
    try:
        run_res = mod.main(variables, name)
        result_queue.put(("success", run_res))
    except Exception as e:
        result_queue.put(("error", str(e)))

# Celery层面的超时只做最后一层兜底，实际超时逻辑自主控制
@app.task(name="execute_script", bind=True, base=CallbackTask, soft_time_limit=330, time_limit=350)
def execute_script(self, script_id, variables, name, start_time):
    self.update_state(state='EXECUTING')
    script = Script.objects.get(pk=script_id)
    mod = Script.get_script(script)

    # 用spawn模式创建子进程，避免fork带来的锁、资源复用问题
    ctx = multiprocessing.get_context("spawn")
    result_queue = ctx.Queue()
    script_proc = ctx.Process(
        target=_script_runner,
        args=(mod, variables, name, result_queue)
    )
    script_proc.start()

    # 自定义超时阈值，单位秒
    exec_timeout = 300
    script_proc.join(timeout=exec_timeout)

    # 超时判定：子进程仍在运行则强制终止
    if script_proc.is_alive():
        script_proc.terminate()
        script_proc.join()
        self.update_state(state='FAILURE', meta={'error': f'脚本执行超时，超过{exec_timeout}秒未完成'})
        raise TimeoutError(f"脚本执行超时，最大允许时长{exec_timeout}秒")

    # 处理子进程返回结果
    if not result_queue.empty():
        status, content = result_queue.get()
        if status == "error":
            raise Exception(content)
    return name

该方案不受脚本逻辑、worker池类型影响，只要超时就会强制杀掉独立的脚本子进程，worker本身完全不受影响，不会出现僵尸进程、worker卡死的问题。

方案3：定时巡检兜底 #

如果存在极端场景漏杀的任务，可以加一层定时巡检做兜底：

• 开启Celery事件发送配置：CELERY_SEND_EVENTS = True
• 配置Celery Beat周期任务，每5分钟扫描一次运行时长超过阈值的execute_script任务，调用app.control.revoke(task_id, terminate=True, signal='SIGKILL')强制终止，同时更新任务状态为失败。

• 不要在Windows环境验证超时逻辑：Windows下Celery的信号处理、进程管理逻辑和Linux差异极大，超时逻辑大概率不生效
• 不要用线程实现超时控制：Python没有强制终止线程的机制，遇到死循环、阻塞C扩展调用时根本无法终止
• 超时终止后记得主动更新任务状态，避免使用django-db后端时任务状态一直卡在EXECUTING

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