如何将函数作为参数传递给芹菜任务？答案

【问题标题】：How to pass function as an argument to a celery task?如何将函数作为参数传递给芹菜任务？
【发布时间】：2021-07-30 20:10:45
【问题描述】：

我想将函数作为参数传递给 celery 任务。我在 stackoverflow 中发现了两个类似的问题（1 和2）。我尝试了答案中提到的解决方案，这就是我目前正在做的事情：

调用者模块内部：

import marshal

def function_to_be_passed:
    # ...

serialized_func_code = marshal.dumps(function_to_be_passed.__code__)
celery_task.delay(serialized_func_code)

在 celery 任务中，我正在反序列化并调用函数：

import marshal, types
from celery.task import task

@task()
def celery_task(serialized_func_code):
    code = marshal.loads(serialized_func_code)
    func = types.FunctionType(code, globals(), "some_func_name")

    # calling the deserialized function
    func()

但是，在调用 celery_task.delay(serialized_func_code) 时，我收到错误消息：

kombu.exceptions.EncodeError: 'utf-8' codec can't decode byte 0xe3 in 位置 0：无效的继续字节

在下面分享堆栈跟踪：

File “...caller.py",
 celery_task.delay(
File "/python3.8/site-packages/celery/app/task.py", line 426, in delay
  return self.apply_async(args, kwargs)
File "/python3.8/site-packages/celery/app/task.py", line 555, in apply_async
  content_type, content_encoding, data = serialization.dumps(
File "/python3.8/site-packages/kombu/serialization.py", line 221, in dumps
  payload = encoder(data)
File "/python3.8/contextlib.py", line 131, in __exit__
  self.gen.throw(type, value, traceback)
File "/python3.8/site-packages/kombu/serialization.py", line 54, in _reraise_errors
  reraise(wrapper, wrapper(exc), sys.exc_info()[2])
File "/python3.8/site-packages/vine/five.py", line 194, in reraise
  raise value.with_traceback(tb)
File "/python3.8/site-packages/kombu/serialization.py", line 50, in _reraise_errors
  yield
File "/python3.8/site-packages/kombu/serialization.py", line 221, in dumps
  payload = encoder(data)
File "/python3.8/site-packages/kombu/utils/json.py", line 69, in dumps
  return _dumps(s, cls=cls or _default_encoder,
File "/python3.8/site-packages/simplejson/__init__.py", line 398, in dumps
  return cls(
File "/python3.8/site-packages/simplejson/encoder.py", line 296, in encode
  chunks = self.iterencode(o, _one_shot=True)
File "/python3.8/site-packages/simplejson/encoder.py", line 378, in iterencode
    return _iterencode(o, 0)
kombu.exceptions.EncodeError: 'utf-8' codec can't decode byte 0xe3 in position 0: invalid continuation byte

当我使用 pickle 提供的dumps 和loads 时出现类似错误。

kombu.exceptions.EncodeError: 'utf-8' codec can't decode byte 0x80 in 位置 0：无效的起始字节

我使用的是 Django 版本 2.2.17 和 Python 版本 3.8.5。

【问题讨论】：

标签： python celery pickle marshalling kombu

【解决方案1】：

可以在here 找到一个好的答案。总之，您可以将任务序列化程序更改为pickle，而不是here 中提到的json。小心不要从外部资源中腌制任何东西，因为这可能很危险。

【讨论】：