如何处理因Python导入模块中C++断言引发的Celery WorkerLostError？

嘿，这个问题我之前踩过坑！C++里的assert()触发时会直接调用abort()终止整个进程，Python的try-except根本拦不住——因为进程都没了，Celery自然会抛出WorkerLostError，而link_error只处理任务本身抛出的Python异常，Worker进程崩溃这种底层问题它管不到。下面给你几个可行的解决方案：

1. 最根本的方法：把C++断言换成Python可捕获的异常 #

既然你用的是pybind11，完全可以把C++里的断言逻辑改成抛出异常，让pybind11自动转换成Python异常，这样Python端就能正常捕获了。

改造C++代码 #

把原来的assert(condition)替换成条件判断+抛出std::runtime_error（或者自定义异常）：

#include <pybind11/pybind11.h>
#include <stdexcept>

namespace py = pybind11;

void risky_cpp_func(int input) {
    // 替换assert(input > 0);
    if (input <= 0) {
        throw std::runtime_error("Input must be a positive integer");
    }
    // 你的业务逻辑
}

PYBIND11_MODULE(my_cpp_module, m) {
    m.def("risky_func", &risky_cpp_func, "Do something with C++");
    // 注册C++异常到Python，pybind11会自动转换
    py::register_exception<std::runtime_error>(m, "RuntimeError");
}

Python端Celery任务处理 #

现在你可以用普通的try-except捕获异常，link_error也能正常触发了：

from celery import task
from my_cpp_module import risky_func

@task(bind=True)
def process_with_cpp(self, input_val):
    try:
        result = risky_func(input_val)
        return {"status": "success", "data": result}
    except RuntimeError as e:
        # 可以选择重试，或者直接标记失败并返回错误信息
        self.update_state(state='FAILURE', meta={"error": str(e)})
        # 抛出异常让Celery记录失败，link_error会收到这个异常
        raise e

2. 无法修改C++模块？用子进程隔离崩溃风险 #

如果这个C++模块是第三方的，没法改代码，那就只能把调用它的逻辑放到独立子进程里执行——这样即使子进程因为断言崩溃，也不会影响Celery Worker进程。

用multiprocessing实现隔离 #

from celery import task
from multiprocessing import Process, Queue
import my_cpp_module

# 子进程执行的函数，负责调用C++模块
def run_cpp_task(input_val, result_queue):
    try:
        output = my_cpp_module.risky_func(input_val)
        result_queue.put(("ok", output))
    except Exception as e:
        result_queue.put(("error", str(e)))

@task(bind=True)
def safe_process_task(self, input_val):
    result_queue = Queue()
    # 启动子进程
    proc = Process(target=run_cpp_task, args=(input_val, result_queue))
    proc.start()
    # 设置超时时间，防止子进程挂起
    proc.join(timeout=60)

    # 检查子进程退出状态：非0说明异常终止（比如触发assert）
    if proc.exitcode != 0:
        error_msg = "C++模块执行崩溃，请检查输入参数"
        self.update_state(state='FAILURE', meta={"error": error_msg})
        raise RuntimeError(error_msg)
    
    # 获取子进程返回结果
    if not result_queue.empty():
        status, data = result_queue.get()
        if status == "ok":
            return {"status": "success", "data": data}
        else:
            raise RuntimeError(data)

3. 配置Celery处理WorkerLostError #

如果上面的方法都没法用，还可以通过Celery的自定义任务基类来捕获WorkerLostError，并做用户通知：

from celery import Task
from celery.exceptions import WorkerLostError

class CppTask(Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        # 检查是否是WorkerLostError
        if isinstance(exc, WorkerLostError):
            # 这里写通知用户的逻辑：比如更新数据库状态、发送邮件/消息
            print(f"任务 {task_id} 因C++模块崩溃失败，请通知用户检查输入")
        super().on_failure(exc, task_id, args, kwargs, einfo)

# 任务使用自定义基类
@task(base=CppTask)
def risky_task(self, input_val):
    return my_cpp_module.risky_func(input_val)

另外记得在Django的Celery配置里加上CELERY_ACKS_LATE = True，这样Worker崩溃后，任务会被重新调度（或者你可以在on_failure里标记任务为不可重试）。

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