2015-07-19 119 views
0

下面的代码片段发起任务,我芹菜安装启动:芹菜和丢失消息

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。

回答

0

同时,我已经更换了以下解决我的问题:与

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

如下:

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