python - asyncio.run（和其他风格）无法从 Celery 中的异常中恢复

Question

我正在使用 Celery 作为零星任务的工人。我需要从我的任务中调用的库都是异步的，我尝试了各种方法来让它工作。

它主要工作，除非我运行的任务引发异常。它很好地记录了异常，但是下次它抓取任务时，它会出错，说“事件循环已关闭”（如果任务正确完成，则不会发生这种情况）。

我尝试了以下口味，结果都一样；

@app.task
def score_survey(sku, auto_push=False):
    async_to_sync(score_survey_handler)(sku, auto_push)

@app.task
def score_survey(sku, auto_push=False):
    asyncio.run(score_survey_handler(sku, auto_push))

@app.task
def score_survey(sku, auto_push=False):
    loop = asyncio.new_event_loop()
    task = loop.create_task(score_survey_handler(sku, auto_push))
    try:
        loop.run_until_complete(task)
    except:
        task.exception()
        task.cancel()
        raise
    finally:
        loop.stop()
        loop.close()

@app.task
def score_survey(sku, auto_push=False):
    loop = asyncio.new_event_loop()
    loop.run_until_complete(score_survey_handler(sku, auto_push))

所有完全相同的行为；只要score_survey_handler完成没有任何问题，它就可以正常工作，但是一旦它抛出异常，Celery 接受的下一个任务就会抛出一个异常，表示事件循环已关闭。

我也试过只使用异常而不是在 Celery 任务中引发它们，这也有效 - 但我确实需要将异常记录到 Celery 结果后端，所以这远非理想。

知道我怎样才能让它工作吗？或者为什么即使我打电话也不起作用new_event_loop()？

标签： pythoncelerypython-asyncio