如何有条件地执行芹菜任务python答案

【问题标题】：how to execute celery tasks conditionally python如何有条件地执行芹菜任务python
【发布时间】：2017-06-18 17:04:57
【问题描述】：

我是芹菜新手。我有一个芹菜任务，需要在满足条件时执行。否则几分钟后重试。从下面的代码中，我被困在如何在其他情况下重试相同的任务？感谢您的帮助。

@app.task(bind=True,soft_time_limit=4 * 3600)
def task_message_queue(id, name=None, tid=None, src=None, dest=None, queue="MessageQueue"):
   ThreadLocalStore().set_data({"id": id, "tid": tid, "name": name,"src": src, "dest":dest})
   num_files = os.popen("find %s -type f | wc -l" % dest).read().strip().split('\n')[0]
   if num_files < 20:
     #Move files from src to destination
   else:
     #wait for 2 minutes and retry the task

【问题讨论】：

标签： python celery celery-task

【解决方案1】：

您必须致电retry 让 celery 重试任务，并且您可以设置倒计时，以便 celery 等待这么长时间并重试任务。下面是从官方celery docs借来的代码。修改@task 根据您的需要装饰器以及self.retry

from celery.task import task
@app.task(bind=True,soft_time_limit=4 * 3600)
def task_message_queue(self, id, name=None, tid=None, src=None, dest=None, queue="MessageQueue"):
   ThreadLocalStore().set_data({"id": id, "tid": tid, "name": name,"src": src, "dest":dest})
   num_files = os.popen("find %s -type f | wc -l" % dest).read().strip().split('\n')[0]
   try:
       if num_files < 20:
           #Move files from src to destination
       else:
           raise SOME_EXCEPTION
           #wait for 2 minutes and retry the task
   except SOME_EXCEPTION as exc:
       self.retry(exc=exc, countdown=TIME_TO_WAIT_BEFORE_RETRY)

【讨论】：

感谢阿皮特。所以，我在 else 条件下添加了一个“raise”。和 self.retry 在异常中。我会测试一下。
其实你不需要加薪条件，如果这里可以工作的话
如果我不引发异常，则未定义“exc”。即使我引发异常，我也遇到了问题，因为 exc 没有执行实际任务

【解决方案2】：

您也可以再次触发任务，而不是依赖重试。

from celery.task import task
@app.task(bind=True,soft_time_limit=4 * 3600)
def task_message_queue(self, id, name=None, tid=None, src=None, dest=None, queue="MessageQueue"):
   ThreadLocalStore().set_data({"id": id, "tid": tid, "name": name,"src": src, "dest":dest})
   num_files = os.popen("find %s -type f | wc -l" % dest).read().strip().split('\n')[0]

   if num_files < 20:
       #Move files from src to destination
   else:
       # Trigger the task again in 120 seconds.
       task_message_queue.apply_async(countdown=120)

【讨论】：