Python使用signal定时结束AsyncIOScheduler任务的问题
作者:返回主页临渊(v:superz-han) 时间:2022-12-19 21:28:11
在使用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() # 永远运行
来源:https://www.cnblogs.com/superhin/p/15060818.html
标签:Python,signal,AsyncIOScheduler,任务
0
投稿
猜你喜欢
用Python实现一个模仿UP主弹幕控制的直播间功能
2023-02-24 18:13:27
Python3.7 新特性之dataclass装饰器
2021-05-11 13:13:40
ORACLE 正则解决初使化数据格式不一致
2009-05-24 19:44:00
Microsoft JET Database Engine 错误 '80040e21' 所有记录中均未找到搜索关键字
2009-08-30 13:07:00
H1标签的定义
2008-07-29 12:43:00
javascript让浏览器实现复读机的功能
2008-10-10 11:49:00
Php多进程实现代码
2023-10-11 09:23:14
在任意字符集下正常显示网页的方法二(续)
2023-11-22 17:36:14
经验总结:修改MySQL默认密码的具体步骤
2009-01-04 13:29:00
JS实现仿Windows经典风格的选项卡Tab切换代码
2023-08-25 05:33:54
python实现矩阵乘法
2023-11-03 07:41:10
用python下载百度文库的代码
2023-07-31 00:40:44
14条最佳JS代码编写技巧[译]
2009-11-09 19:37:00
python turtle绘图命令及案例
2022-04-29 10:26:58
FrontPage XP设计制作网页小技巧八则
2008-06-04 12:43:00
为MySQL提权简单方法
2009-08-29 15:21:00
PHP脚本监控Nginx 502错误并自动重启php-fpm
2023-09-26 00:11:45
JS中的public和private对象,即static修饰符
2023-08-29 21:53:13
PHP扩展之kafka安装应用案例详解
2023-09-06 09:53:43
JavaScript 日期下拉选择器
2008-10-31 12:13:00