Python中用asyncio实现每隔N秒执行函数且不替换run_forever循环

实现方案 #

有两种完全基于asyncio原生能力的实现方式，均不会阻塞事件循环中其他正在运行的任务，无需替换原有run_forever逻辑。

方案1：用loop.call_later递归调度（适合do_something为普通同步函数的场景）

call_later是asyncio事件循环原生提供的延时调度能力，在回调函数执行完成后重新注册下一次调度，即可实现周期性执行：

import asyncio

def do_something():
    # 你的业务逻辑
    print("执行do_something")
    # 执行完成后注册下一次10秒后的调度
    loop.call_later(10, do_something)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    # 首次注册：启动后10秒第一次执行，如果需要启动立刻执行，直接先调用一次do_something()即可
    loop.call_later(10, do_something)
    loop.run_forever()

方案2：用异步任务实现（适合do_something可以改造为协程的场景，灵活度更高）

asyncio的异步sleep是完全非阻塞的，事件循环会在sleep期间调度其他任务运行：

import asyncio

async def do_something():
    # 你的业务逻辑，若有同步耗时操作建议用run_in_executor扔到线程池执行，避免阻塞事件循环
    print("执行do_something")

async def schedule_task():
    while True:
        await do_something()
        # 非阻塞等待10秒，不会影响事件循环其他任务运行
        await asyncio.sleep(10)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    # 创建异步任务提交到事件循环
    loop.create_task(schedule_task())
    loop.run_forever()

补充说明 #

以上两种方案默认的间隔是「上一次函数执行完成」到「下一次函数开始执行」的间隔为10秒，如果需要严格按照固定10秒间隔触发（忽略函数自身执行耗时），可以自行增加时间计算逻辑调整sleep时长即可。
若你的do_something包含耗时同步操作，建议用loop.run_in_executor包裹调用，避免同步逻辑阻塞整个事件循环：

await loop.run_in_executor(None, do_something)

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