Celery任务传递动态生成类实例遇序列化问题求助

替代实现方案 #

如果上述修复无法生效，可通过以下方式绕过动态类序列化问题：

1. 传递类元数据，Worker端重建实例

不直接传递实例，而是传递生成动态类所需的元数据，在Worker端重新创建类并实例化：

# 任务发送端
def send_task():
    base_class_path = "myapp.models.BaseClass"
    dynamic_class_attrs = {"custom_attr": "value"}
    my_task.delay(base_class_path, dynamic_class_attrs)

# Celery任务
@app.task
def my_task(base_class_path, dynamic_class_attrs):
    from importlib import import_module
    # 导入基类
    module_name, class_name = base_class_path.rsplit(".", 1)
    base_class = getattr(import_module(module_name), class_name)
    # 重建动态类并实例化
    child_class = type("ChildClass", (base_class,), dynamic_class_attrs)
    instance = child_class()
    # 执行业务逻辑

2. 提取实例状态为普通可序列化对象

将动态类实例的关键状态提取为字典或JSON可序列化结构，Worker端拿到后重新构建实例：

# 任务发送端
instance = ChildClassGenerator.generate(BaseClass)
# 提取实例状态（根据类结构自定义）
instance_state = {
    "base_class": "myapp.models.BaseClass",
    "attrs": instance.__dict__
}
my_task.delay(instance_state)

# Celery任务
@app.task
def my_task(instance_state):
    from importlib import import_module
    # 导入基类并重建实例
    module_name, class_name = instance_state["base_class"].rsplit(".", 1)
    base_class = getattr(import_module(module_name), class_name)
    child_class = type("ChildClass", (base_class,), {})
    instance = child_class()
    instance.__dict__.update(instance_state["attrs"])
    # 执行业务逻辑

3. 使用单进程Worker池（Solo Pool）

如果性能要求不高，可使用Solo池（单进程模式），避免进程间对象序列化：

celery -A your_app worker --loglevel=info --pool=solo

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