Luigi多异步调用工作流问题：事件循环已关闭RuntimeError

问题根源 #

Luigi是同步调度框架，而MS Graph SDK的接口为异步实现。如果在多个Task或异步调用中重复创建并关闭事件循环（比如多次调用asyncio.run()），前一次调用关闭循环后，后续异步操作再试图使用已关闭的循环就会触发该错误。你的临时方案（合并为单个async函数）能生效，是因为只创建/关闭了一次循环，避免了循环生命周期冲突。

正确的异步调用管理方案 #

1. 复用全局事件循环实例 #

不要在每个异步操作里单独创建循环，而是在工作流初始化时创建一个全局循环，所有Task的异步逻辑都复用这个循环，直到整个工作流结束再关闭。

示例代码：

import asyncio
import luigi
from msgraph.core import GraphClient

# 全局事件循环，仅初始化一次
global_loop = asyncio.new_event_loop()

class MSGraphBaseTask(luigi.Task):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.graph_client = GraphClient(credential=your_credential)
        self.loop = global_loop

    def run(self):
        # 所有异步逻辑通过这个循环执行，不会中途关闭
        self.loop.run_until_complete(self.async_run())

    async def async_run(self):
        # 子类实现具体异步逻辑
        pass

2. 拆分多步逻辑并批量管理异步任务 #

保留「获取-提取-删除」的分步职责，用asyncio.gather()或create_task把多个异步任务放到同一个循环上下文里执行，确保循环在所有任务完成后再处理。

示例分步Task：

import json

class FetchDriveItemTask(MSGraphBaseTask):
    item_id = luigi.Parameter()

    async def async_run(self):
        # 获取驱动器项
        self.item = await self.graph_client.get(f"/drives/items/{self.item_id}")
        # 保存结果供后续Task使用
        self.output().open('w').write(self.item.json())

    def output(self):
        return luigi.LocalTarget(f"fetch_{self.item_id}.json")

class ExtractDriveContentTask(MSGraphBaseTask):
    item_id = luigi.Parameter()

    def requires(self):
        return FetchDriveItemTask(self.item_id)

    async def async_run(self):
        item_data = json.load(self.input().open())
        # 提取内容
        content = await self.graph_client.get(item_data['@microsoft.graph.downloadUrl'])
        self.output().open('wb').write(content.content)

    def output(self):
        return luigi.LocalTarget(f"content_{self.item_id}.bin")

class DeleteDriveItemTask(MSGraphBaseTask):
    item_id = luigi.Parameter()

    def requires(self):
        return ExtractDriveContentTask(self.item_id)

    async def async_run(self):
        # 删除驱动器项
        await self.graph_client.delete(f"/drives/items/{self.item_id}")

# 工作流入口
class OneDriveWorkflow(luigi.Task):
    item_id = luigi.Parameter()

    def requires(self):
        return DeleteDriveItemTask(self.item_id)

    def run(self):
        # 工作流结束后统一关闭全局循环（可选，若Luigi进程结束则自动回收）
        global_loop.close()

3. 避免循环嵌套与自动关闭 #

不要在异步函数内部使用asyncio.run()，它会自动创建并关闭循环，导致外部循环失效。所有异步逻辑都用await，并依托全局循环执行。

4. 集成测试的额外处理 #

集成测试中如果涉及多Task调度，要确保全局循环在测试开始前初始化，测试结束后再关闭，避免测试用例间的循环状态污染。

为什么临时方案是「次优」的 #

合并为单个async函数虽然能解决错误，但把「获取-提取-删除」三个耦合度低的逻辑硬绑在一起，违反了Luigi的Task拆分原则（每个Task做单一职责，便于重试、调度和监控）。上面的方案既保留了Task的独立性，又解决了事件循环的生命周期问题。

内容的提问来源于stack exchange，提问作者Harper Marchman-Jones