我想将客户创建的函数传递给celery任务并执行。例如下面的示例,我正在尝试编写一个
可能函数没有正确序列化(可以理解,这很难)。但是,有没有任何方法可以做到这一点?最佳实践是什么?我想我可以传递一个字符串,然后
编辑:我发现了一种序列化函数的方法...我想我可以将其包装起来完成我需要做的事情。还有更好的想法吗?
然后,我尝试使用这个任务与这个:
在 worker 上,我得到了这个:
map
函数,该函数接受函数f
和列表l
并在celery任务中执行map(f, l)
。可能函数没有正确序列化(可以理解,这很难)。但是,有没有任何方法可以做到这一点?最佳实践是什么?我想我可以传递一个字符串,然后
exec
它,但是我的应用程序意图不是这样的。编辑:我发现了一种序列化函数的方法...我想我可以将其包装起来完成我需要做的事情。还有更好的想法吗?
from celery import Celery
app = Celery('tasks', broker='redis://localhost', backend='redis://localhost')
@app.task
def cp_map(f, l):
return map(f, l)
然后,我尝试使用这个任务与这个:
In [20]: from tasks import cp_map
In [21]: def f(x): return x + 1
In [22]: cp_map.delay(f, [1,2,3])
Out[22]: <AsyncResult: 27baf8bf-8ef3-496d-a445-ebd7ee94e206>
In [23]: _.status
Out[23]: 'PENDING'
在 worker 上,我得到了这个:
[2014-02-09 22:27:00,828: CRITICAL/MainProcess] Can't decode message body: DecodeError(AttributeError("'module' object has no attribute 'f'",),) (type:u'application/x-python-serialize' encoding:u'binary' raw:"'\\x80\\x02}q\\x01(U\\x07expiresq\\x02NU\\x03utcq\\x03\\x88U\\x04argsq\\x04c__main__\\nf\\nq\\x05]q\\x06(K\\x01K\\x02K\\x03e\\x86q\\x07U\\x05chordq\\x08NU\\tcallbacksq\\tNU\\x08errbacksq\\nNU\\x07tasksetq\\x0bNU\\x02idq\\x0cU$27baf8bf-8ef3-496d-a445-ebd7ee94e206q\\rU\\x07retriesq\\x0eK\\x00U\\x04taskq\\x0fU\\x0ctasks.cp_mapq\\x10U\\ttimelimitq\\x11NN\\x86U\\x03etaq\\x12NU\\x06kwargsq\\x13}q\\x14u.' (233b)"')
Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/kombu/messaging.py", line 585, in _receive_callback
decoded = None if on_m else message.decode()
File "/usr/local/lib/python2.7/dist-packages/kombu/message.py", line 142, in decode
self.content_encoding, accept=self.accept)
File "/usr/local/lib/python2.7/dist-packages/kombu/serialization.py", line 184, in loads
return decode(data)
File "/usr/lib/python2.7/contextlib.py", line 35, in __exit__
self.gen.throw(type, value, traceback)
File "/usr/local/lib/python2.7/dist-packages/kombu/serialization.py", line 59, in _reraise_errors
reraise(wrapper, wrapper(exc), sys.exc_info()[2])
File "/usr/local/lib/python2.7/dist-packages/kombu/serialization.py", line 55, in _reraise_errors
yield
File "/usr/local/lib/python2.7/dist-packages/kombu/serialization.py", line 184, in loads
return decode(data)
File "/usr/local/lib/python2.7/dist-packages/kombu/serialization.py", line 64, in pickle_loads
return load(BytesIO(s))
DecodeError: 'module' object has no attribute 'f'
kombu.exceptions.EncodeError: 'utf-8' codec can't decode byte 0xe3 in position 0: invalid continuation byte
- Harris Irfan