为Django项目中的可重用应用程序创建基于类的Celery任务

Question

对于Django项目来说，创建基于函数的任务是非常干净的。只需在django应用程序中创建tasks.py，然后开始编写任务，如以下示例所示，该示例取自http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html

from __future__ import absolute_import, unicode_literals
from celery import shared_task
@shared_task
def add(x, y):
return x + y

@shared_task
def mul(x, y):
return x * y

但有时基于功能的任务是紧密耦合的，并且不太可重用。所以我想创建一个基于类的芹菜任务，它被记录在官方网站上。以下内容之后https://github.com/celery/celery/issues/3874我可以创建示例任务，但我不确定它是否是创建基于类的任务的正确方法。

from __future__ import absolute_import, unicode_literals
from celery import shared_task, Task
import time
from celery import current_app

@shared_task
def add(x, y):
time.sleep(5)
return x + y

@shared_task
def mul(x, y):
return x * y
# Sample class based task for testing
class AnotherTask(current_app.Task):
name = 'tasks.another_task'
def __init__(self, x, y):
self.x = x
self.y = y
def run(self):
time.sleep(5)
return self.x + self.y
# We need to manually register this class based task    
current_app.tasks.register(AnotherTask(3, 4))

我可以调用此任务，但每个调用结果值都是相同的

(workflow) alok@alok-MacBookAir:~/exp/expdjango/mysite$ ./manage.py shell
Python 3.6.3 (default, Oct  3 2017, 21:45:48) 
Type "copyright", "credits" or "license" for more information.
IPython 5.1.0 -- An enhanced Interactive Python.
?         -> Introduction and overview of IPython's features.
%quickref -> Quick reference.
help      -> Python's own help system.
object?   -> Details about 'object', use 'object??' for extra details.
In [1]: from polls.tasks import *
In [2]: r = AnotherTask(3,4).delay()
In [3]: r.get()
Out[3]: 7
In [4]: r = AnotherTask(5,6).delay()
In [5]: r.get()
Out[5]: 7

这是创建和调用基于类的任务的正确方法吗？

Answer 1

基于类的任务实际上在每个运行时只实例化一次。如果您希望您的任务被参数化，请向run方法添加参数：

class AnotherTask(current_app.Task):
name = 'tasks.another_task'
def run(self, x, y):
time.sleep(5)
return x + y

然后你可以这样称呼它：

r = AnotherTask().delay(3, 4)
# Or
r = AnotherTask().apply_async(args=[3, 4])

这也在基于类的任务的文档中进行了描述，尤其是在实例化部分。