52

我正在尝试将类的方法用作 django-celery 任务，并使用 @task 装饰器对其进行标记。Anand Jeyahar提出了同样的情况。是这样的

class A:
    @task
    def foo(self, bar):
        ...

def main():
    a = A()
    ...
    # what i need
    a.foo.delay(bar) # executes as celery task 
    a.foo(bar) # executes locally

问题是即使我像这样使用a.foo.delay(bar)它所说的类实例，也foo需要至少两个参数，这意味着self指针丢失。

更多信息：

  • 由于继承，我无法将类转换为模块
  • 方法强烈依赖于类成员，所以我不能让它们成为静态的
  • 使用@task 装饰器将类标记为任务使该类本身成为任务，并且可以从方法执行run()方法，使用一些参数作为方法选择的键，但这并不是我想要的。
  • 创建一个类的实例并将其作为self参数传递给方法会改变我执行方法的方式，而不是像芹菜一样，而是像通常的方法一样（即在测试时）
  • 我试图找出如何从构造函数中动态地注册任务，但是 celery 在工作人员之间共享代码，所以这就是为什么它似乎是不可能的。

谢谢你的帮助！

4

5 回答 5

53

Celery 自 3.0 版起实验性地支持将方法用作任务。

这方面的文档在 中celery.contrib.methods，并且还提到了一些您应该注意的警告：

https://docs.celeryproject.org/en/3.1/reference/celery.contrib.methods.html

注意：从4.0 起支持contrib.methods从 Celery 中移除

于 2012-02-13T13:36:08.063 回答
10

Jeremy Satterfield 有一个简洁直接的教程来编写基于类的任务，如果这是你想要完成的。你可以在这里查看。

魔术基本上是扩展celery.Task类，包括一个run()方法，就像这样：

from celery import Task

class CustomTask(Task):
    ignore_result = True

    def __init__(self, arg):
        self.arg = arg

    def run(self):
        do_something_with_arg(self.arg)

然后像这样运行任务：

your_arg = 3

custom_task = CustomTask()
custom_task.delay(your_arg)

我不确定ignore_result = True零件是否必要。

于 2015-04-02T22:26:01.033 回答
4

当你有：

    a = A()

你可以做：

    A.foo.delay(a, param0, .., paramN)

干杯

于 2012-07-03T18:22:25.083 回答
2

我遇到了类似的情况，并决定将类方法包装在一个简单的函数中，该函数会将其参数重定向到类的实例，并执行此类方法：

class A:
    def foo(self, bar):
       # do this

a = A()

@app.task
def a_wrapper(bar):
    return a.foo(bar)

# probably in a different size with an import in-place:

a_wrapper.delay(bar)
于 2020-11-11T08:06:18.077 回答
1

对我来说，唯一有效的是 celery.current_app 因为这只是传递self给方法。

所以这应该是这样的：

from celery import current_app
from celery.contrib.methods import task_method

class A:
@current_app.task(filter=task_method, name='A.foo')
def foo(self, bar):
    ...

如果您在不同的类中有同名的方法，则必须使用该名称。

于 2014-08-27T13:44:53.237 回答