芹菜和丢失的信息

Question

下面的代码片段在我的芹菜安装中启动任务:

tasks.py :

@app.task(ignore_result=False)
def asyncTransactionTask(txid):
    Here I do something with txid and do not schedule additional tasks
@app.task(ignore_result=True)
def asyncCheckNotifications(*args):
    try:
        payments = # get an array of values
        payments_tasks = []
        for payment in payments:
            payments_tasks.append(asyncTransactionTask.s(payment))
        chain(group(payments_tasks) | asyncCheckNotifications.subtask()).apply_async(countdown=60)
    except Exception as e:
        logger.error(str(e))
        asyncCheckNotifications.apply_async(countdown=10)
        raise e
asyncCheckNotifications.delay()

我期望看到asyncCheckNotifications方法大约每分钟运行一次，然而我每两分钟收到一次。

更重要的是，如果我检查计划任务(celery -A myapp inspect scheduled)，我看到方法执行被适当地安排，但是当我达到超时时，它只是被下一分钟的另一个计划所取代，并且什么都没有运行。

我正在使用芹菜3.1.8。消息代理是RabbitMQ 3.2.4.

Answer 1

同时，我通过替换以下内容解决了我的问题:

chain(group(payments_tasks) | asyncCheckNotifications.subtask()).apply_async(countdown=60)

与以下内容:

    chain(group(payments_tasks) | asyncCheckNotifications.subtask(countdown=60)).delay()