41

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

例如我有这个任务：

@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”。

4

4 回答 4

91

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

更多细节在这里：http ://docs.celeryproject.org/en/master/userguide/canvas.html#immutability

于 2012-12-09T14:51:37.047 回答
3

一种可能的解决方案已经发布，但我想进一步澄清和替代解决方案（在某些情况下是更好的解决方案）。

您看到的错误表明您的任务签名需要考虑第二个参数，这是由于在 a 中调用任务时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)

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

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

于 2016-10-10T08:33:51.590 回答
-1

你可以尝试做这样的事情。您可以有两个参数，而不是函数 tprint 的单个参数

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

然后

chain(tprint.s('a', 0) | tprint.s('b'))()
于 2012-12-03T05:08:30.930 回答
-2

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

我不知道 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。

于 2012-12-04T09:56:35.223 回答