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

对于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

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