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

问题描述

我想将一个函数作为参数传递给 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' 编解码器无法解码字节 0xe3 位置 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",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' 编解码器无法解码字节 0x80 位置 0：无效的起始字节

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

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）