基于以下示例:芹菜4.0.0和基于任务的任务工作流程
http://shulhi.com/class-based-celery-task/;
https://blog.balthazar-rouberol.com/celery-best-practices
http://shulhi.com/class-based-celery-task/
Class based Task in django celery
我想创建类似的东西:
class CalculationWorker(Task):
def __init__(self, id_user):
self._id_user = id_user
self._user = get_object_or_404(User, pk=self._id_user)
# I need to understand if the bind work or if it's needed
def _bind(self, app):
return super(self.__class__, self).bind(celery_app)
def _retrieve_some_task(self):
# long calculation
def _long_run_task(self):
# long calculation
self._retrieve_some_task()
# Main entry
def run(self):
self._long_run_task()
#
def run_job():
worker = CalculationWorker(id_user=323232)
task = worker.apply_async()
的文件似乎说这是可能的(反正这是我不清楚)http://docs.celeryproject.org/en/latest/userguide/tasks.html#custom-task-classes 。它甚至说:
“”” 这意味着初始化构造仅会被每个进程调用,该任务类在语义上更接近演员 。‘’
但http://docs.celeryproject.org/en/latest/whatsnew-4.0.html#the-task-base-class-no-longer-automatically-register-tasks显式说:“最好的做法是使用自定义任务类只用于覆盖一般行为,然后使用任务装饰来实现任务”
至于结果,我得到了一个NotRegistered例外由于这种https://github.com/celery/celery/issues/3548,但加入app.tasks.register(CalculationWorker())
没解决不了。 我正在使用Django 1.10.X和芹菜4.0.0
这种方法仍然有效吗?
感谢