如何使用AsyncIOScheduler從另一個函式運行函式？

我很難理解它是如何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功能無效。

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函式沒有運行，我怎樣才能將它們放在一起回圈？

uj5u.com熱心網友回復：

我想你想main()永遠運行你的功能，所以你可以用這種方式嘗試。

controllers.py

import asyncio
from datetime import datetime


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


async def bar():
    while True:
        print(f'{datetime.now()} Bar')
        await asyncio.sleep(1)

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()

    await asyncio.create_task(bar())


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像這樣。

標籤：

上一篇：如何使用discord.pybot從另一個python檔案發送訊息？

下一篇：SpringBoot自動提交手動事務，無需呼叫存盤庫