在每个 Celery Task 运行时做一些事情答案

【问题标题】：Do something on every Celery Task run在每个 Celery Task 运行时做一些事情
【发布时间】：2016-07-06 15:44:23
【问题描述】：

假设我有一些基于一个抽象任务的 Celery 任务。

class BaseTask(Task):
    abstract = True

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logging.info('Task failed')

    def on_success(self, retval, task_id, args, kwargs):
        logging.info('Task success')

    # NO SUCH METHOD IN TASK
    # def on_start(self):
    #     do_something_on_every_task_start()

@app.task(base=BaseTask)
def task1(x):
    print x

@app.task(base=BaseTask)
def task2(y):
    print y

我想在每个任务执行开始时做一些事情。有没有可能达到这个结果？谢谢。

【问题讨论】：

标签： python celery

【解决方案1】：

您可以覆盖__call__ 方法。如果您使用task1(1) 同步调用任务或使用task1.delay(1) 异步调用它，这将起作用。在以下示例中，您将获得“CALLED！”在控制台或日志中，具体取决于您调用任务的方式以及日志记录的设置方式。

class BaseTask(Task):
    abstract = True

    def __call__(self, *args, **kwargs):
        print "CALLED!"
        return super(BaseTask, self).__call__(*args, **kwargs)

@app.task(base=BaseTask)
def task1(x):
    print x

@app.task(base=BaseTask)
def task2(y):
    print y

【讨论】：

好的，谢谢您的回答。我认为这是正确的，但我想知道为什么 celery 项目没有提供语义上更漂亮的方法来做到这一点......

【解决方案2】：

如果需要“语义优美”的方法，可以试试信号。

http://docs.celeryproject.org/en/latest/userguide/signals.html

这是另一篇关于前/后任务回调的有用帖子。

https://gist.github.com/nuria/35f332acfb84ecf80e3b

【讨论】：