芹菜链不成批工作

Question

乍一看，我非常喜欢芹菜中的“批”特性，因为在调用API之前，我需要分组一定数量的I(否则我可能会被赶出去)。

不幸的是，当测试一点点时，批处理任务似乎不能很好地处理其他的画布原语，在本例中是链。例如：

@a.task(base=Batches, flush_every=10, flush_interval=5)
def get_price(requests):
    for request in requests:
        a.backend.mark_as_done(request.id, 42, request=request)
        print "filter_by_price " + str([r.args[0] for r in requests])

@a.task
def completed():
    print("complete")

因此，使用这个简单的工作流：

chain(get_price.s("ID_1"), completed.si()).delay()

我看到这个输出：

[2015-07-11 16:16:20,348: INFO/MainProcess] Connected to redis://localhost:6379/0
[2015-07-11 16:16:20,376: INFO/MainProcess] mingle: searching for neighbors
[2015-07-11 16:16:21,406: INFO/MainProcess] mingle: all alone
[2015-07-11 16:16:21,449: WARNING/MainProcess] celery@ultra ready.
[2015-07-11 16:16:34,093: WARNING/Worker-4] filter_by_price ['ID_1']

5秒后，filter_by_price()就会像预期的那样被触发。问题是completed()从未被调用过。

对这里会发生什么有什么想法吗？如果不使用批处理，有什么合适的方法来解决这个问题呢？

PS:，我已经设置了CELERYD_PREFETCH_MULTIPLIER=0，就像文档说的那样。

Answer 1

看起来，批处理任务的行为与正常任务有很大的不同。批处理任务甚至没有发出像success这样的信号。

因为您需要在completed之后调用get_price任务，所以可以直接从get_price本身调用它。

@a.task(base=Batches, flush_every=10, flush_interval=5)
def get_price(requests):
    for request in requests:
         # do something
    completed.delay()