• 售前

  • 售后

热门帖子
入门百科

Python使用signal定时竣事AsyncIOScheduler任务的题目

[复制链接]
123456811 显示全部楼层 发表于 2021-8-14 12:58:24 |阅读模式 打印 上一主题 下一主题
在使用aiohttp联合apscheduler的AsyncIOScheduler模仿定点并发的时间遇到两个题目
  1. <ol>
  2.     <li>在调度器scheduler.start()后,程序直接退出(在Jupiter中任务可以正常启动)</li>
  3.     <li>如何在指定时间调用scheduler.shutdown()? (因为程序直接退出了)
  4.     </li>
  5. </ol>
复制代码
原调试代码如下:
  1. from datetime import datetime, timedelta
  2. import aiohttp
  3. from apscheduler.schedulers.asyncio import AsyncIOScheduler
  4. from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
  5. async def get(session):
  6.     url = 'https://httpbin.org/get?a=1'
  7.     async with session.get(url) as res:
  8.         print('get', res.status)
  9.         return await res.text()
  10. async def post(session):
  11.     url = 'https://httpbin.org/post?b=2'
  12.     async with session.post(url) as res:
  13.         print('post', res.status)
  14.         return await res.text()
  15. async def main():
  16.     async with aiohttp.ClientSession() as session:
  17.         await get(session)
  18.         await post(session)
  19. if __name__ == '__main__':
  20.     jobstores = {'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}
  21.     scheduler = AsyncIOScheduler(jobstores=jobstores)
  22.     for i in range(10):  # 添加10个任务
  23.         job = scheduler.add_job(main, 'date', run_date=datetime.now() + timedelta(seconds=10))
  24.     scheduler.start()
复制代码
Google后发现AsyncIOScheduler的使用须要在scheduler启动后,须要自己调用
  1. asyncio.get_event_loop().run_forever()
复制代码
来启动协程使命。
但是一旦run_forever()则就会阻塞至死。除非有KeyboardInterrupt, SystemExit等非常大概强杀来停止其运行。
此时想到使用Python的signal来定时发送信号,修改后步伐如下,可以正常耽误停止(感觉有点像模仿Go的defer)。
  1. # -*- coding: utf-8 -*-
  2. """
  3. @Time : 2021/7/23
  4. @Auth : hanzhichao
  5. @Desc:
  6. """
  7. from datetime import datetime, timedelta
  8. import signal
  9. import asyncio
  10. import aiohttp
  11. from apscheduler.schedulers.asyncio import AsyncIOScheduler
  12. from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
  13. async def get(session):
  14.     url = 'https://httpbin.org/get?a=1'
  15.     async with session.get(url) as res:
  16.         print('get', res.status)
  17.         return await res.text()
  18. async def post(session):
  19.     url = 'https://httpbin.org/post?b=2'
  20.     async with session.post(url) as res:
  21.         print('post', res.status)
  22.         return await res.text()
  23. async def main():
  24.     async with aiohttp.ClientSession() as session:
  25.         await get(session)
  26.         await post(session)
  27. if __name__ == '__main__':
  28.     jobstores = {'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}
  29.     scheduler = AsyncIOScheduler(jobstores=jobstores)
  30.     for i in range(10):  # 添加10个任务
  31.         job = scheduler.add_job(main, 'date', run_date=datetime.now() + timedelta(seconds=10))
  32.     scheduler.start()
  33.     signal.alarm(20)  # 20秒后终止程序
  34.     asyncio.get_event_loop().run_forever()  # 永远运行
复制代码
到此这篇关于Python使用signal定时结束AsyncIOScheduler使命的文章就先容到这了,更多干系Python定时结束AsyncIOScheduler使命内容请搜索脚本之家从前的文章或继续浏览下面的干系文章盼望各人以后多多支持脚本之家!

帖子地址: 

回复

使用道具 举报

分享
推广
火星云矿 | 预约S19Pro,享500抵1000!
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

草根技术分享(草根吧)是全球知名中文IT技术交流平台,创建于2021年,包含原创博客、精品问答、职业培训、技术社区、资源下载等产品服务,提供原创、优质、完整内容的专业IT技术开发社区。
  • 官方手机版

  • 微信公众号

  • 商务合作