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

Posted 2023-02-24

【中文标题】在 Django 项目中为可重用应用程序创建基于类的 Celery 任务

【英文标题】：Create class based Celery task for reusable app in Django project

【发布时间】：2018-12-24 21:20:55

【问题描述】：

对于 Django 项目来说，创建基于函数的任务是相当干净的。只需在 django 应用程序中创建 tasks.py 并开始编写类似此示例的任务，该示例取自 http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html 的官方 celery 文档

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

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

【参考方案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])

这也在the docs on class based tasks 中进行了描述，特别是在instantiation section 中。