新型芹菜api的问题

I have a class that extends celerys Task. It runs just fine with the old style API, but I am having problems converting it to the new API.# In app/tasks.pyfrom celery import Celery, Taskcelery = Celery()@celery.taskclass CustomTask(Task): def run(self, x): try: # do something except Exception, e: self.retry(args=[x], exc=e)And then I run the task like so -CustomTask().apply_async(args=[x], queue='q1')And I get the error -TypeError: run() takes exactly 2 arguments (1 given)This SO answer seems to do the same thing and it was accepted so presumably it works. Can anyone help me out and explain to me why my code isn't working?EDITThis works if I name the task, different from the class name -name = 'app.tasks.CustomTask2'But if I keep the name of the task the same as the full class name, it doesn't workname = 'app.tasks.CustomTask'But the problem with having a different name is that celery has an extra task, with the same name as the task class name. 解决方案 The decorator is not used with classes, it's used for functions.Usually you will not want to define custom task classes unless you want to implement common behavior.So either remove the @celery.task decorator, or use it with a function.Note that the task you define here is not bound with any celery instanceNote bound to any specific app instance:from celery import Taskclass MyTask(Task): passYou can bind it later:from celery import Celeryapp = Celery(broker='amqp://')MyTask.bind(app)or you can use the base class available on the app:from celery import Celeryapp = Celery(broker='amqp://')class MyTask(app.Task): passThe last example is not very clean as it means you are finalizing the app at module level,this is another reason why using the task decorator with functions is a best practice, and only create custom classes to be used as a base class for decorated tasks (@task(base=MyTask)). 这篇关于新型芹菜api的问题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！