Python使用signal定时结束AsyncIOScheduler任务的问题_Python

Python使用signal定时结束AsyncIOScheduler任务的问题

2021-12-15 10:32临渊(v:superz-han) Python

这篇文章主要介绍了Python使用signal定时结束AsyncIOScheduler任务,在使用aiohttp结合apscheduler的AsyncIOScheduler模拟定点并发的时候遇到两个问题，针对每个问题给大家详细介绍，需要的朋友可以参考下

在使用aiohttp结合apscheduler的AsyncIOScheduler模拟定点并发的时候遇到两个问题

在调度器scheduler.start()后，程序直接退出（在Jupiter中任务可以正常启动）

如何在指定时间调用scheduler.shutdown()? (因为程序直接退出了）

原调试代码如下：

				?

									from datetime import datetime, timedelta

									import aiohttp

									from apscheduler.schedulers.asyncio import AsyncIOScheduler

									from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

									async 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个任务

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

									    scheduler.start()

Google后发现AsyncIOScheduler的使用需要在scheduler启动后，需要自己调用asyncio.get_event_loop().run_forever()来启动协程任务。
但是一旦run_forever()则就会阻塞至死。除非有KeyboardInterrupt, SystemExit等异常或者强杀来停止其运行。
此时想到使用Python的signal来定时发送信号，修改后程序如下，可以正常延迟停止（感觉有点像模拟Go的defer)。

				?

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

									"""

									@Time ： 2021/7/23

									@Auth ： hanzhichao

									@Desc：

									"""

									from datetime import datetime, timedelta

									import signal

									import asyncio

									import aiohttp

									from apscheduler.schedulers.asyncio import AsyncIOScheduler

									from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

									async 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个任务

									        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()  # 永远运行