第Python使用signal定時結(jié)束AsyncIOScheduler任務(wù)的問題fromapscheduler.schedulers.asyncioimportAsyncIOScheduler

fromtores.sqlalchemyimportSQLAlchemyJobStore

asyncdefget(session):

url='/geta=1'

asyncwithsession.get(url)asres:

print('get',res.status)

returnawaitres.text()

asyncdefpost(session):

url='/postb=2'

asyncwithsession.post(url)asres:

print('post',res.status)

returnawaitres.text()

asyncdefmain():

asyncwithaiohttp.ClientSession()assession:

awaitget(session)

awaitpost(session)

if__name__=='__main__':

jobstores={'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}

scheduler=AsyncIOScheduler(jobstores=jobstores)

foriinrange(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等異?；蛘邚?qiáng)殺來停止其運(yùn)行。

此時想到使用Python的signal來定時發(fā)送信號，修改后程序如下，可以正常延遲停止（感覺有點像模擬Go的defer)。

#-*-coding:utf-8-*-

@Time：2025/7/23

@Auth：hanzhichao

@Desc：

fromdatetimeimportdatetime,timedelta

importsignal

importasyncio

importaiohttp

fromapscheduler.schedulers.asyncioimportAsyncIOScheduler

fromtores.sqlalchemyimportSQLAlchemyJobStore

asyncdefget(session):

url='/geta=1'

asyncwithsession.get(url)asres:

print('get',res.status)

returnawaitres.text()

asyncdefpost(session):

url='/postb=2'

asyncwithsession.post(url)asres:

print('post',res.status)

returnawaitres.text()

asyncdefmain():

asyncwithaiohttp.ClientSession()assession:

awaitget(session)

awaitpost(session)

if__name__=='__main__':

jobstores={'default':SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}

scheduler=AsyncIOScheduler(jobstores=jobstores)

foriinrange(10):#添加10個任務(wù)

job=scheduler.add_job(main,'date',run_date=datetime.now()+timedelta(seconds