Python使用signal定时结束AsyncIOScheduler任务的问题
【Python使用signal定时结束AsyncIOScheduler任务的问题】在使用aiohttp结合apscheduler的AsyncIOScheduler模拟定点并发的时候遇到两个问题
原调试代码如下:
- 在调度器scheduler.start()后,程序直接退出(在Jupiter中任务可以正常启动)
- 如何在指定时间调用scheduler.shutdown()? (因为程序直接退出了)
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个任务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, 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个任务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()# 永远运行
到此这篇关于Python使用signal定时结束AsyncIOScheduler任务的文章就介绍到这了,更多相关Python定时结束AsyncIOScheduler任务内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!
推荐阅读
- 由浅入深理解AOP
- 【译】20个更有效地使用谷歌搜索的技巧
- mybatisplus如何在xml的连表查询中使用queryWrapper
- MybatisPlus|MybatisPlus LambdaQueryWrapper使用int默认值的坑及解决
- MybatisPlus使用queryWrapper如何实现复杂查询
- python学习之|python学习之 实现QQ自动发送消息
- 逻辑回归的理解与python示例
- python自定义封装带颜色的logging模块
- 【Leetcode/Python】001-Two|【Leetcode/Python】001-Two Sum
- iOS中的Block