如何将带有对象参数的对象传递给 celery 任务？答案

【问题标题】：How to pass an object with an object parameter to a celery task?如何将带有对象参数的对象传递给 celery 任务？
【发布时间】：2021-12-31 06:51:46
【问题描述】：

我有一个对象model_test_api，如下所示。

import test_script
model_test_api = test_script.TestAPI(api)

需要新TestAPI的api也是一个对象。

import model_api
api = model_api.Model_API()

我正在尝试以下two methods 将model_test_api 传递给celery 任务。

1. another_function.apply_async(model_test_api)
2. another_function.apply_async(lambda: test_script.TestAPI(api))

但是，他们都给了我一个错误kombu.exceptions.EncodeError: Object of type function is not JSON serializable。然后我将 model_test_api 打包到一个 json 对象中。

import json
json_test_api = json.dumps(model_test_api.__dict__)
another_function.apply_async(json_test_api)

它给了我另一个错误TypeError: Object of type Model_API is not JSON serializable。任何人都知道我该怎么做？提前致谢。

【问题讨论】：

python 中的一切都是对象。您正在传递它，但正如错误消息明确指出的那样，您传递的东西不是 JSON 可序列化的，这可能是必需的。
错误消息告诉我们您的描述没有告诉我们的事情：您正在尝试创建 JSON。但是，只有您的示例 2 正在传递一个函数。示例 1 是传递一个对象。但是，对象也不能转换为 JSON。您从哪里获得要运行的代码？
Thanks@TimRoberts here another_function 这里描述的是celery 的分布式函数。
TestAPI(api) 返回什么？它真的是对象构造器还是装饰器？
@jasonshu 一切都是对象。说“它返回一个对象”并没有告诉我们任何事情。

标签： python python-3.x celery

【解决方案1】：

在回答问题之前先介绍一些内容，因此 celery 为 python 启用了事件流。您有一个数据库，其中包含需要处理的任务的queue，通常这个数据库是postgres 或redis。有工作人员正在观察队列，当有东西被添加到队列中时，网络上具有可用处理资源的工作人员会选择任务，执行处理，然后将该任务标记为已完成。

因此，当您触发another_function.apply_async 行时，这意味着名为another_function 的任务被添加到队列中，并且传入的任何参数值也被添加。这意味着所有参数都需要序列化才能存储在数据库postgres 或redis 中。 Celery 无法将 python 对象存储在队列中的内存中，因为它可能是执行处理的完全不同的机器，并且它的 RAM 中没有对象。

我将给出一个推荐实现的示例，假设您有一个名为CIA_Agent_Model 的模型，它具有三个属性，一个主键和一些文本字段。您有一个名为 another_function 的 celery 任务，并且您想传入 CIA_Agent_Model 实例，您可以使用 another_function.delay(agent_model_instance.id) 调用 celery 任务，其中传入的参数是对象的主键，而不是对象本身。然后在 celery 任务中，您再次通过 ID 从数据库中获取对象。也可以序列化模型实例https://www.django-rest-framework.org/api-guide/serializers/，序列化结果可以传入another_function.delay(model_serialisation_result)。

您可以创建一个自定义函数，将 test_script.TestAPI(api) 的返回对象转换为 json dict，然后将该 json dict 传递给 celery 任务调用参数。

【讨论】：