FastAPI实现任务启停/暂停/恢复API遇Resume异常求助

问题描述 #

实现了一个基于asyncio事件的FastAPI接口，支持run、pause、resume、abort操作控制后台任务，但调用resume接口时抛出如下异常：

RuntimeError("Task <Task pending name='Task-4' coro=<TaskManager.async_task() running at>> got Future attached to a different loop")

复现步骤：

• 启动服务器
• 调用/run接口启动任务
• 调用/pause接口暂停任务
• 调用/resume接口恢复任务

原因分析 #

问题出在asyncio同步原语（Event）的循环绑定特性：asyncio.Event对象会绑定到创建它时所在的事件循环。你的代码中TaskManager实例是在uvicorn.run()之前创建的，此时的事件循环是主线程的默认循环；而uvicorn启动后会创建自己的事件循环来处理请求和任务，导致任务中等待的_pause_event属于另一个循环，跨循环操作就会触发该异常。

解决方案 #

将TaskManager的初始化延迟到FastAPI的启动事件中，确保_pause_event是在uvicorn运行的事件循环里创建的。修改后的代码如下：

import asyncio
from fastapi import FastAPI
import uvicorn

app = FastAPI()

task_manager = None

class TaskManager:
    def __init__(self):
        self._pause_event = asyncio.Event()
        self._task_running = False
        self._current_task = None

    async def async_task(self):
        while True:
            await self._pause_event.wait()
            print("Async Task is running...")
            # Simulate some asynchronous work
            await asyncio.sleep(2)

    async def start_task(self):
        if not self._task_running:
            self._task_running = True
            self._pause_event.set()
            self._current_task = asyncio.create_task(self.async_task())
            print("Task started.")
        else:
            print("Task is already running.")

    async def pause_task(self):
        self._task_running = False
        self._pause_event.clear()
        print("Task paused.")

    async def resume_task(self):
        self._task_running = True
        self._pause_event.set()
        print("Task resumed.")

    async def abort_task(self):
        self._task_running = False
        self._pause_event.set()
        if self._current_task:
            self._current_task.cancel()
            try:
                await self._current_task
            except asyncio.CancelledError:
                print("Task aborted successfully.")
        print("Task aborted.")

@app.on_event("startup")
async def startup_event():
    global task_manager
    task_manager = TaskManager()

@app.post("/run")
async def run_background_task():
    await task_manager.start_task()
    return {"message": "Task started."}

@app.post("/pause")
async def pause_background_task():
    await task_manager.pause_task()
    return {"message": "Task paused."}

@app.post("/resume")
async def resume_background_task():
    await task_manager.resume_task()
    return {"message": "Task resumed."}

@app.post("/abort")
async def abort_background_task():
    await task_manager.abort_task()
    return {"message": "Task aborted."}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=5049)

另外补充完善了abort_task方法，确保任务被取消后能正确处理CancelledError，避免未捕获的异常。

关键修改点 #

1. 将task_manager改为全局变量，初始化为None
2. 通过FastAPI的startup事件初始化TaskManager，此时的asyncio.Event会绑定到uvicorn运行的事件循环
3. 完善abort_task的错误处理逻辑

内容的提问来源于stack exchange，提问作者aikiOr