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

Question

我有一个应用程序已经运行无限地使用 asyncio 事件循环运行并且我还需要每 10 秒运行一个特定的函数。

def do_something():
   pass

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

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

已编辑：我可以用下面的代码实现这个

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

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

Answer 1

您可以通过

实现

async def do_something():
   while True:
      await asyncio.wait(10)
      ...rest of code...


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(do_something())
    loop.run_forever()

Answer 2

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 本身是协程或长运行ning 子例程，则设计必须略有不同。在前一种情况下使用 await func(...)，在后一种情况下使用 asyncio's thread capabilities.

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

Run a function every n seconds in python with asyncio

python

python-asyncio

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

Run a function every n seconds in python with asyncio

python

python-asyncio

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