如何使用 AsyncIOScheduler 从另一个函数运行函数？

Question

我很难理解它是如何AsyncIOScheduler工作的以及如何在主函数内安排一个函数，或者如何是正确的方法来做到这一点。

如何foo每 10 秒运行一次该函数？

假设我有这样的结构：

package/
    controllers.py
    main.py

从文件中controllers.py我调用了一个函数foo，该函数如下所示：

async def foo():
    print('Hello World')

我想foo从文件运行该函数（以及许多其他函数）main.py，该文件如下：

import asyncio
from controllers import foo, bar
from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def main():

    # Init message
    print('\nPress Ctrl-C to quit at anytime!\n' )
    
    await asyncio.create_task(bar())

    scheduler = AsyncIOScheduler()
    scheduler.add_job(await asyncio.create_task(foo()), "interval", seconds=10)
    scheduler.start()

if __name__ == "__main__":
    while True:
        try:
            asyncio.run(main())
        except Exception as e:
            print("Exception: " + str(e))

这样运行调度程序是否正确？或者每次main函数运行时定时器都会被重置？

我尝试了下面的代码，循环间隔有效，但该main功能无效。

import asyncio
from controllers import foo, bar
from apscheduler.schedulers.asyncio import AsyncIOScheduler

async def main():

    # Init message
    print('\nPress Ctrl-C to quit at anytime!\n' )
    
    await asyncio.create_task(bar())

if __name__ == "__main__":

    scheduler = AsyncIOScheduler()
    scheduler.add_job(foo, "interval", seconds=10)
    scheduler.start()

    while True:
        try:
            asyncio.get_event_loop().run_forever()
            asyncio.run(main())
        except Exception as e:
            print("Exception: " + str(e))

如果我将订单更改为：

            asyncio.get_event_loop().run_forever()
            asyncio.run(main())

到：

            asyncio.run(main())
            asyncio.get_event_loop().run_forever()

我收到错误：There is no current event loop in thread 'MainThread'.

循环有效（仅限调度程序），但该main函数未运行，如何将它们放在循环中？

Answer 1

我想你想main()永远运行你的函数，所以你可以尝试这种方式。

controllers.py

from datetime import datetime


async def foo():
    print(f'{datetime.now()} Foo')


async def bar():
    print(f'{datetime.now()} Bar')

main.py

import asyncio
from controllers import foo, bar
from apscheduler.schedulers.asyncio import AsyncIOScheduler


async def main():
    # Init message
    print('\nPress Ctrl-C to quit at anytime!\n')

    scheduler = AsyncIOScheduler()
    scheduler.add_job(foo, "interval", seconds=2)
    scheduler.start()
    
    while True:
        await asyncio.create_task(bar())
        await asyncio.sleep(1)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.create_task(main())
    loop.run_forever()

输出：

Press Ctrl-C to quit at anytime!
2022-01-09 03:38:10.468914 Bar
2022-01-09 03:38:11.472966 Bar
2022-01-09 03:38:12.468882 Foo
2022-01-09 03:38:12.473449 Bar
2022-01-09 03:38:13.477678 Bar
2022-01-09 03:38:14.469389 Foo
2022-01-09 03:38:14.480098 Bar
2022-01-09 03:38:15.483889 Bar
2022-01-09 03:38:16.472584 Foo
2022-01-09 03:38:16.487105 Bar
2022-01-09 03:38:17.492550 Bar
2022-01-09 03:38:18.467403 Foo
2022-01-09 03:38:18.493330 Bar
2022-01-09 03:38:19.496680 Bar
2022-01-09 03:38:20.471553 Foo
2022-01-09 03:38:20.499928 Bar
2022-01-09 03:38:21.503572 Bar
2022-01-09 03:38:22.470329 Foo
2022-01-09 03:38:22.505597 Bar
2022-01-09 03:38:23.510854 Bar
2022-01-09 03:38:24.471455 Foo
...

此外，您可以运行周期函数而不使用AsyncIOScheduler像这样的。