如何重新调度asyncio中call_later安排的协程？以资源延迟关闭为例

首先明确：asyncio 中call_later返回的TimerHandle并没有直接重新调度的API，所以取消旧任务再创建新任务是标准实现方式，但可以通过封装逻辑让代码更简洁，避免重复的取消/创建代码。

方案1：封装延迟任务工具类 #

把取消和重新调度的逻辑封装到一个类里，业务代码只需调用统一的方法：

from fastapi import FastAPI, Depends
import asyncio

class DelayedTask:
    def __init__(self, loop, delay, callback):
        self.loop = loop
        self.delay = delay
        self.callback = callback
        self._handle = None

    def schedule(self):
        # 取消已有任务（如果存在），再创建新任务
        if self._handle is not None:
            self._handle.cancel()
        self._handle = self.loop.call_later(self.delay, self._run_callback)

    def _run_callback(self):
        self._handle = None
        self.callback()

def create_app():
    app = FastAPI()
    app._my_resource = None
    app._delayed_close_task = None

    def _close_my_resource():
        print("closing resource")
        app._my_resource = None

    async def my_resource():
        loop = asyncio.get_running_loop()
        if app._my_resource is None:
            print("opening resource")
            app._my_resource = "Hello World"
            # 初始化延迟关闭任务
            app._delayed_close_task = DelayedTask(loop, 5, _close_my_resource)
        
        # 只需调用schedule，内部自动处理取消和重新调度
        app._delayed_close_task.schedule()

        return app._my_resource
        
    @app.get("/")
    def index(res = Depends(my_resource)):
        return {"my_resource": res}

    return app

方案2：用异步任务实现（适合异步关闭逻辑） #

如果资源关闭操作本身是异步的（比如关闭摄像头需要异步IO），可以用asyncio.create_task结合asyncio.sleep实现，逻辑更贴合异步场景：

from fastapi import FastAPI, Depends
import asyncio

def create_app():
    app = FastAPI()
    app._my_resource = None
    app._close_task = None

    async def _close_my_resource():
        await asyncio.sleep(5)
        print("closing resource")
        app._my_resource = None
        app._close_task = None

    async def my_resource():
        if app._my_resource is None:
            print("opening resource")
            app._my_resource = "Hello World"
        
        # 取消旧的关闭任务并处理CancelledError
        if app._close_task is not None:
            app._close_task.cancel()
            try:
                await app._close_task
            except asyncio.CancelledError:
                pass
        
        # 创建新的延迟关闭任务
        app._close_task = asyncio.create_task(_close_my_resource())

        return app._my_resource
        
    @app.get("/")
    def index(res = Depends(my_resource)):
        return {"my_resource": res}

    return app

总结 #

两种方案本质都是基于「取消旧任务+创建新任务」的逻辑，但通过封装可以让业务代码更简洁，避免重复冗余的操作。如果是同步的关闭逻辑，方案1更轻量；如果是异步关闭逻辑，方案2更合适。

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