django-celery:动态创建并注册任务

Question

我想用延迟添加任意任务django-celery.目前,我已经创建了一个类似于下面的类(只是一个例子,实际的类有更多):

from celery.task import task

class Delayer(object):
    def delay(self, func, minutes):
        return task(func, name="%s.delayed"%self.__class__.__name__)\
            .apply_async(countdown=minutes*60)

我正在运行芹菜如下:

python manage.py celeryd -E -B -lDEBUG

当我尝试从django shell [例如Delayer().delay(lambda: 1, 1)]中运行我的延迟方法时,我在我的celeryd输出中遇到这样的错误:

[2013-01-02 15:26:39,324: ERROR/MainProcess] Received unregistered task of type "Delayer.delayed".
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you are using relative imports?
Please see http://bit.ly/gLye1c for more information.

The full contents of the message body was:
{'retries': 0, 'task': "Delayer.delayed", 'eta': '2013-01-02T21:27:39.320913', 'args': [], 'expires': None, 'callbacks': None, 'errbacks': None, 'kwargs': {}, 'id': '99d49fa7-bd4b-40b0-80dc-57309a6f19b1', 'utc': True} (229b)

Traceback (most recent call last):
  File "/home/simon/websites/envs/delayer/local/lib/python2.7/site-packages/celery/worker/consumer.py", line 432, in on_task_received
    strategies[name](message, body, message.ack_log_error)
KeyError: "Delayer.delayed"

我的问题是,是否可以注册这些动态创建的任务？如果没有,我可以用什么方法来使用芹菜来达到同样的效果？

Answer 1

简单的答案是你做不到; 因为芹菜在不同的进程中运行,它需要能够导入任何作为芹菜任务运行的代码; 你生成的callable不是,所以celery移动引用callables的方法不起作用.

然而,这确实提出了一种可能的攻击方式:如果你能想出一种不同的序列化你的可调用方法,那么你可以将它作为一个简单的芹菜任务的参数.此前的问题可能有所帮助.请注意安全警示:-)