在中注册基于 class 的任务

Question

我正在使用 Celery 4.0.2 版。

与以前版本的 Celery 相比，基于 class 的任务似乎不会自动注册（即，如果您配置了自动发现）。

但是，我什至没有实现手动注册基于 class 的任务。

根据 Celery 更改日志：

http://docs.celeryproject.org/en/latest/changelog.html#version-4-0-1

从 4.0.1 版本开始，应该可以手动注册任务：

from celery import Celery, Task
app = Celery()

class CustomTask(Task):

    def run(self):
        return 'hello'

app.register_task(CustomTask())

但这似乎不起作用。有谁知道如何做到这一点？

我尝试了一些正在讨论的建议（除了集成 https://github.com/celery/celery/issues/3744 中提到的自定义任务加载器）：

https://github.com/celery/celery/issues/3615

https://github.com/celery/celery/issues/3744

Answer 1

快到了！您需要在注册的任务上调用 delay()。

这可行：

from celery import Celery, Task

app = Celery()


class CustomTask(Task):
    def run(self):
        return 'hello'


task = CustomTask()
app.register_task(task)

task.delay()

Answer 2

如果你需要shared_task装饰器：

from celery import Task, shared_task

class CustomTask(Task):
    def process(self):
        return 'hello'

@shared_task(bind=True, base=CustomTask)
def custom(self):
    self.process()

process是启动任务的自定义名称（装饰器覆盖run方法）

bind=True 将函数绑定到 class 实例

base=CustomTask 为任务设置基础 class

在中注册基于 class 的任务

Register class-based task in

python

celery

celery-task