Celery：如何忽略弦或链中的任务结果？

Question

我正在使用celery，我有几个任务需要按顺序执行。

例如我有这个任务：

@celery.task
def tprint(word):
    print word

我想做这样的事情：

>>> chain(tprint.s('a') | tprint.s('b'))()

然后我得到

TypeError: tprint() takes exactly 1 argument (2 given)

。

与和弦相同，在这种情况下，我需要在一组任务之后执行一个任务：

>>> chord([tprint.s('a'), tprint.s('b')])(tprint.s('c'))

那么遇到这种情况该如何处理呢？我不关心每个任务的结果，但它们需要按顺序执行。

---

添加第二个参数不起作用：

@celery.task
def tprint(word, ignore=None):
    print word

>>> chain(tprint.s('a', 0) | tprint.s('b'))()

这将打印出“a”和“None”。

Answer 1

有一个内置功能可以忽略链接和其他结果 - 不可变的子任务。您可以使用 .si() 快捷方式代替 .s() 或 .subtask(immutable=True)

更多详细信息请参见：https://docs.celeryq.dev/en/stable/userguide/canvas.html#immutability

Answer 2

已经发布了一个可能的解决方案，但我想添加进一步的说明和替代解决方案（在某些情况下是更好的解决方案）。

您看到的错误表明您的任务签名需要考虑第二个参数，这是因为在

chain

中调用任务时，Celery 会自动将每个任务

result

作为第一个参数推送的以下任务。

来自文档：

任务可以链接在一起，这实际上意味着添加回调任务：

>>> res = add.apply_async((2, 2), link=mul.s(16))
>>> res.get()
4

将应用链接任务，并将其父任务的结果作为第一个参数

因此，就您的情况而言，您可以像这样重写您的任务：

@celery.task
def tprint(result, word):
    print word

如果您不打算对结果做任何事情，您也可以通过更改装饰器来忽略它：

@celery.task(ignore_result=True)

然后您就不必更改任务签名。

抱歉，最后一点需要进一步研究。

Answer 3

你可以尝试做这样的事情。 函数 tprint 可以有 2 个参数，而不是只有一个参数

def tprint(word, x=None):
    print word

然后

chain(tprint.s('a', 0) | tprint.s('b'))()

Answer 4

最终找到解决方法，链式装饰器将完成这项工作。

我不知道 celery 到底是如何做到的，但 celery 似乎强制将上一个任务的结果绑定到下一个任务的第一个参数。

这是一个例子：

def chain_deco(func):

    @functools.wraps(func)
    def wrapper(chain=None, *args, **kwargs):
        if chain is False:
            return False

        func(*args, **kwargs)
        return True

    return wrapper


@celery.task
@chain_deco
def hello(word):
    print "hello %s" % word

现在这将给出正确的输出。

>>> (hello.s(word='a') | hello.s(word='b'))()

或

>>> (hello.s('a') | hello.s('b'))(True)

并且装饰器还提供了在中间停止一条链的能力（使后面的级联失败。）

同样的机制也应该适用于

chord

。