检查是否在芹菜任务中答案

【问题标题】：Check if in celery task检查是否在芹菜任务中
【发布时间】：2016-03-22 15:51:46
【问题描述】：

如何检查celery执行的函数？

def notification():
   # in_celery() returns True if called from celery_test(), 
   #                     False if called from not_celery_test()
   if in_celery():
      # Send mail directly without creation of additional celery subtask
      ...
   else:
      # Send mail with creation of celery task
      ...

@celery.task()
def celery_test():
    notification()

def not_celery_test():
    notification()

【问题讨论】：

标签： python flask celery celery-task

【解决方案1】：

这是使用celery.current_task 的一种方法。这是任务要使用的代码：

def notification():
    from celery import current_task
    if not current_task:
        print "directly called"
    elif current_task.request.id is None:
        print "called synchronously"
    else:
        print "dispatched"

@app.task
def notify():
    notification()

这是您可以运行以执行上述操作的代码：

        from core.tasks import notify, notification
        print "DIRECT"
        notification()
        print "NOT DISPATCHED"
        notify()
        print "DISPATCHED"
        notify.delay().get()

我在第一个 sn-p 中的任务代码位于名为 core.tasks 的模块中。我将最后一个 sn-p 中的代码推送到自定义 Django 管理命令中。这测试了 3 个案例：

直接拨打notification。
通过同步执行的任务调用notification。也就是说，这个任务不是通过 Celery 分派给工人的。该任务的代码在调用notify的同一进程中执行。
通过工作人员运行的任务调用notification。任务的代码在与启动它的进程不同的进程中执行。

输出是：

NOT DISPATCHED
called synchronously
DISPATCHED
DIRECT
directly called

DISPATCHED 之后的输出任务中没有来自print 的行，因为该行最终出现在工作日志中：

[2015-12-17 07:23:57,527: WARNING/Worker-4] dispatched

重要提示：我最初在第一次测试中使用if current_task is None，但它不起作用。我检查并重新检查。不知何故，Celery 将current_task 设置为一个看起来像None 的对象（如果你在上面使用repr，你会得到None）但不是None。不确定那里发生了什么。使用if not current_task 有效。

另外，我已经在 Django 应用程序中测试了上面的代码，但我没有在生产中使用它。可能有一些我不知道的问题。

【讨论】：