使用 asyncio 在 python 中每 n 秒运行一个函数

Question

我有一个应用程序，它已经无限运行，异步事件循环永远运行，而且我需要每 10 秒运行一个特定的函数。

def do_something():
   pass

a = asyncio.get_event_loop()
a.run_forever()

我想每 10 秒调用一次函数 do_something 。如何在不使用 while 循环替换 asynctio 事件循环的情况下实现这一目标？

编辑：我可以用下面的代码来实现这一点

def do_something():
   pass
while True:
   time.sleep(10)
   do_something()

但我不想使用 while 循环在我的应用程序中无限运行，而是想使用 asyncio run_forever()。那么如何使用 asyncio 每 10 秒调用相同的函数呢？有没有类似的调度程序不会阻止我正在进行的工作？

Answer 1

asyncio不附带内置调度程序，但构建您自己的调度程序很容易。只需将while循环与asyncio.sleep每隔几秒运行一次代码结合起来即可。

async def every(__seconds: float, func, *args, **kwargs):
    while True:
        func(*args, **kwargs)
        await asyncio.sleep(__seconds)

a = asyncio.get_event_loop()
a.create_task(every(1, print, "Hello World"))
...
a.run_forever()

func请注意，如果本身是协程或长时间运行的子例程，则设计必须略有不同。在前一种情况下使用线程功能await func(...)，在后一种情况下使用asyncio线程功能。