检查是否在芹菜任务

Question

如何检查芹菜执行的功能？

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()

Answer 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()

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

• notification直接打电话.

• 调用notification同步执行的任务.也就是说,这项任务不会通过Celery发送给工人.任务代码在调用的同一进程中执行notify.

• 调用notification工作人员执行的任务.任务代码在与启动它的进程不同的进程中执行.

输出是:

NOT DISPATCHED
called synchronously
DISPATCHED
DIRECT
directly called

print输出后的任务中没有行,DISPATCHED因为该行最终在工作日志中:

[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应用程序中测试了上面的代码,但我没有在生产中使用它.可能有些我不知道的问题.