在使用aiohttp結(jié)合apscheduler的AsyncIOScheduler模擬定點并發(fā)的時候遇到兩個問題

在調(diào)度器scheduler.start()后，程序直接退出（在Jupiter中任務(wù)可以正常啟動） 如何在指定時間調(diào)用scheduler.shutdown()? (因為程序直接退出了）

原調(diào)試代碼如下：

from datetime import datetime, timedeltaimport aiohttpfrom apscheduler.schedulers.asyncio import AsyncIOSchedulerfrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStoreasync def get(session): url = ’https://httpbin.org/get?a=1’ async with session.get(url) as res:print(’get’, res.status)return await res.text()async def post(session): url = ’https://httpbin.org/post?b=2’ async with session.post(url) as res:print(’post’, res.status)return await res.text()async def main(): async with aiohttp.ClientSession() as session:await get(session)await post(session)if __name__ == ’__main__’: jobstores = {’default’: SQLAlchemyJobStore(url=’sqlite:///jobs.sqlite’)} scheduler = AsyncIOScheduler(jobstores=jobstores) for i in range(10): # 添加10個任務(wù)job = scheduler.add_job(main, ’date’, run_date=datetime.now() + timedelta(seconds=10)) scheduler.start()

Google后發(fā)現(xiàn)AsyncIOScheduler的使用需要在scheduler啟動后，需要自己調(diào)用asyncio.get_event_loop().run_forever()來啟動協(xié)程任務(wù)。但是一旦run_forever()則就會阻塞至死。除非有KeyboardInterrupt, SystemExit等異常或者強殺來停止其運行。此時想到使用Python的signal來定時發(fā)送信號，修改后程序如下，可以正常延遲停止（感覺有點像模擬Go的defer)。

# -*- coding: utf-8 -*-'''@Time ： 2021/7/23@Auth ： hanzhichao@Desc：'''from datetime import datetime, timedeltaimport signalimport asyncioimport aiohttpfrom apscheduler.schedulers.asyncio import AsyncIOSchedulerfrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStoreasync def get(session): url = ’https://httpbin.org/get?a=1’ async with session.get(url) as res:print(’get’, res.status)return await res.text()async def post(session): url = ’https://httpbin.org/post?b=2’ async with session.post(url) as res:print(’post’, res.status)return await res.text()async def main(): async with aiohttp.ClientSession() as session:await get(session)await post(session)if __name__ == ’__main__’: jobstores = {’default’: SQLAlchemyJobStore(url=’sqlite:///jobs.sqlite’)} scheduler = AsyncIOScheduler(jobstores=jobstores) for i in range(10): # 添加10個任務(wù)job = scheduler.add_job(main, ’date’, run_date=datetime.now() + timedelta(seconds=10)) scheduler.start() signal.alarm(20) # 20秒后終止程序 asyncio.get_event_loop().run_forever() # 永遠(yuǎn)運行

到此這篇關(guān)于Python使用signal定時結(jié)束AsyncIOScheduler任務(wù)的文章就介紹到這了,更多相關(guān)Python定時結(jié)束AsyncIOScheduler任務(wù)內(nèi)容請搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！