25

我有一个定期执行的 check_orders 任务。它创建一组任务，以便我可以计算执行任务所花费的时间，并在它们全部完成后执行某些操作（这是 res.join [1] 和 grouped_subs 的目的）分组的任务是成对的链式任务。

我想要的是当第一个任务不满足条件（失败）时不执行链中的第二个任务。我一生都无法弄清楚这一点，我觉得这对于作业队列管理器来说是非常基本的功能。当我尝试我在 [2] 之后注释掉的东西时（引发异常，删除回调）......我们由于某种原因卡在 check_orders 中的 join() 上（它破坏了组）。对于所有这些任务，我也尝试将 ignore_result 设置为 False，但它仍然不起作用。

@task(ignore_result=True)
def check_orders():
    # check all the orders and send out appropriate notifications
    grouped_subs = []

    for thingy in things:
       ...

        grouped_subs.append(chain(is_room_open.subtask((args_sub_1, )), 
                        notify.subtask((args_sub_2, ), immutable=True)))

    res = group(grouped_subs).apply_async()

    res.join()         #[1]
    logger.info('Done checking orders at %s' % current_task.request.id))

@task(ignore_result=True)
def is_room_open(args_sub_1):
    #something time consuming
    if http_req_and_parse(args_sub_1):
        # go on and do the notify task
        return True
    else:
        # [2]
        # STOP THE CHAIN SOMEHOW! Don't execute the rest of the chain, how?
        # None of the following things work:
        # is_room_open.update_state(state='FAILURE')
        # raise celery.exceptions.Ignore()
        # raise Exception('spam', 'eggs')
        # current_task.request.callbacks[:] = []

@task(ignore_result=True)
def notify(args_sub_2):
    # something else time consuming, only do this if the first part of the chain 
    # passed a test (the chained tasks before this were 'successful'
    notify_user(args_sub_2)
4

3 回答 3

19

在我看来，这是一个常见的用例，在文档中没有得到足够的关注。

假设您想在中途中止一条链，同时仍将 SUCCESS 报告为已完成任务的状态，并且不发送任何错误日志或诸如此类的东西（否则您可以只引发异常），那么实现此目的的一种方法是：

@app.task(bind=True)  # Note that we need bind=True for self to work
def task1(self, other_args):
    #do_stuff
    if end_chain:
        self.request.callbacks = None
        return
    #Other stuff to do if end_chain is False

所以在你的例子中：

@app.task(ignore_result=True, bind=True)
def is_room_open(self, args_sub_1):
    #something time consuming
    if http_req_and_parse(args_sub_1):
        # go on and do the notify task
        return True
    else:
        self.request.callbacks = None

将工作。请注意，您可以使用@abbasov-alexander 所述的快捷方式来代替ignore_result=Trueandsubtask().si()

正如@PhilipGarnero 在评论中所建议的那样，编辑为使用 EAGER 模式。

于 2014-01-14T05:23:50.247 回答
9

这令人难以置信，因为任何官方文档都没有处理如此常见的案例。我不得不处理同样的问题（但使用shared_taskswithbind选项，所以我们可以看到self对象），所以我编写了一个自定义装饰器来自动处理撤销：

def revoke_chain_authority(a_shared_task):
    """
    @see: https://gist.github.com/bloudermilk/2173940
    @param a_shared_task: a @shared_task(bind=True) celery function.
    @return:
    """
    @wraps(a_shared_task)
    def inner(self, *args, **kwargs):
        try:
            return a_shared_task(self, *args, **kwargs)
        except RevokeChainRequested, e:
            # Drop subsequent tasks in chain (if not EAGER mode)
            if self.request.callbacks:
                self.request.callbacks[:] = []
            return e.return_value

    return inner

您可以按如下方式使用它：

@shared_task(bind=True)
@revoke_chain_authority
def apply_fetching_decision(self, latitude, longitude):
    #...

    if condition:
        raise RevokeChainRequested(False)

请参阅此处的完整说明。希望能帮助到你！

于 2014-07-01T20:45:28.833 回答
3

首先，似乎函数中存在异常ignore_result对您没有帮助。

其次，你使用immutable=True意味着下一个函数（在我们的例子中是notify）不接受额外的参数。notify.subtask((args_sub_2, ), immutable=False)如果它适合您的决定，您当然应该使用它。

第三，您可以使用快捷方式：

notify.si(args_sub_2)反而notify.subtask((args_sub_2, ), immutable=True)

和

is_room_open.s(args_sub_1)反而is_room_open.subtask((args_sub_1, ))

尝试使用它的代码：

@task
def check_orders():
    # check all the orders and send out appropriate notifications
    grouped_subs = []

    for thingy in things:
       ...

        grouped_subs.append(chain(is_room_open.s(args_sub_1), 
                                  notify.s(args_sub_2)))

    res = group(grouped_subs).apply_async()

    res.join()         #[1]
    logger.info('Done checking orders at %s' % current_task.request.id))

@task
def is_room_open(args_sub_1):
    #something time consuming
    if http_req_and_parse(args_sub_1):
        # go on and do the notify task
        return True
    else:
        # [2]
        # STOP THE CHAIN SOMEHOW! Don't execute the rest of the chain, how?
        # None of the following things work:
        # is_room_open.update_state(state='FAILURE')
        # raise celery.exceptions.Ignore()
        # raise Exception('spam', 'eggs')
        # current_task.request.callbacks[:] = []
        return False

@task
def notify(result, args_sub_2):
    if result:
        # something else time consuming, only do this if the first part of the chain 
        # passed a test (the chained tasks before this were 'successful'
        notify_user(args_sub_2)
        return True
    return False

如果要捕获异常，则必须像这样使用回调

is_room_open.s(args_sub_1, link_error=log_error.s()) 

from proj.celery import celery

@celery.task
def log_error(task_id):
    result = celery.AsyncResult(task_id)
    result.get(propagate=False)  # make sure result written.
    with open(os.path.join('/var/errors', task_id), 'a') as fh:
        fh.write('--\n\n%s %s %s' % (
            task_id, result.result, result.traceback))
于 2013-07-06T14:18:16.450 回答