Django Celery AbortableTask 用法答案

【问题标题】：Django Celery AbortableTask usageDjango Celery AbortableTask 用法
【发布时间】：2011-04-09 05:49:29
【问题描述】：

我正在尝试使用 Celery 的 AbortableTask 功能，但文档示例似乎对我不起作用。给出的例子是：

from celery.contrib.abortable import AbortableTask

def MyLongRunningTask(AbortableTask):

    def run(self, **kwargs):
        logger = self.get_logger(**kwargs)
        results = []
        for x in xrange(100):
            # Check after every 5 loops..
            if x % 5 == 0:  # alternatively, check when some timer is due
                if self.is_aborted(**kwargs):
                    # Respect the aborted status and terminate
                    # gracefully
                    logger.warning("Task aborted.")
                    return None
            y = do_something_expensive(x)
            results.append(y)
        logger.info("Task finished.")
        return results

和

from myproject.tasks import MyLongRunningTask

def myview(request):

    async_result = MyLongRunningTask.delay()
    # async_result is of type AbortableAsyncResult

    # After 10 seconds, abort the task
    time.sleep(10)
    async_result.abort()

    ...

但是，我得到了错误：

TypeError: MyLongRunningTask() takes exactly 1 argument (0 given)

我做错了什么？

【问题讨论】：

标签： python django celery celery-task

【解决方案1】：

只是猜测，但我认为应该是

class MyLongRunningTask(AbortableTask)

而不是

def MyLongRunningTask(AbortableTask)

【讨论】：

我试了一下（有道理），但错误仍然存在。我一定不能正确地调用任务。
我又试了一次，因为它看起来是对的，而且现在可以了。我假设我只是没有用任务更新文件。