redis - celery 和丢失的消息

Question

以下代码片段启动了我的 Celery 安装中的任务:

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)，我会看到方法执行已适当安排，但当我到达超时时，它只是被另一个替换安排下一分钟，但没有任何运行。

我正在使用 Celery 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()