在Quart/asyncio环境下如何实现不阻塞事件循环的周期性函数调用

你之前的实现存在两个核心错误：

1. asyncio.run() 会创建独立的新事件循环，和Quart服务运行的主事件循环不属于同一个实例，如果你将这段代码放在Quart启动逻辑中，asyncio.run 会阻塞当前线程直到内部协程执行完毕，自然会导致Web服务启动被阻塞。
2. 你的schedule协程仅调度了一次延迟回调，协程本身执行结束后asyncio.run会直接销毁它创建的事件循环，注册的延迟回调还没到执行时间就被清理，所以看不到对应输出。

正确的实现应该利用Quart的服务生命周期钩子，将周期任务挂载到Quart的主事件循环中作为后台任务运行，完全不会阻塞正常的请求处理，示例代码如下：

from quart import Quart
import asyncio
import time

app = Quart(__name__)

# 你的业务执行函数
def do_work():
    print(f"WORK 执行时间: {time.strftime('%Y-%m-%d %H:%M:%S')}", flush=True)

# 周期调度协程，默认间隔60秒执行一次
async def periodic_task(interval: int = 60):
    while True:
        # 若do_work本身是async异步函数，直接await do_work()即可
        # 同步函数用to_thread扔到线程池执行，避免阻塞事件循环
        await asyncio.to_thread(do_work)
        # 等待指定间隔后执行下一次
        await asyncio.sleep(interval)

# Quart服务启动前的钩子，服务启动时自动触发
@app.before_serving
async def init_periodic_task():
    # 用create_task将周期任务扔到事件循环后台调度，不要加await，否则会阻塞服务启动
    asyncio.create_task(periodic_task())

# 正常编写业务路由即可
@app.route('/')
async def index():
    return "服务正常运行"

if __name__ == '__main__':
    app.run()

注意事项 #

• 如果需要严格准点每分钟执行（而不是上次执行结束后间隔60秒），可以在每次执行完计算当前时间到下一分钟的秒数，将asyncio.sleep(interval)替换为对应差值即可
• 如果业务函数执行时间可能超过间隔时间，可以根据需求添加互斥锁，避免同一时间有多个任务实例并行执行

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