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

 更新时间:2021年07月26日 14:20:08   作者:返回主页临渊(v:superz-han)  
这篇文章主要介绍了Python使用signal定时结束AsyncIOScheduler任务,在使用aiohttp结合apscheduler的AsyncIOScheduler模拟定点并发的时候遇到两个问题,针对每个问题给大家详细介绍,需要的朋友可以参考下

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

  1. 在调度器scheduler.start()后,程序直接退出(在Jupiter中任务可以正常启动)
  2. 如何在指定时间调用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()  # 永远运行

到此这篇关于Python使用signal定时结束AsyncIOScheduler任务的文章就介绍到这了,更多相关Python定时结束AsyncIOScheduler任务内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • 利用Python实现批量加密Excel文件

    利用Python实现批量加密Excel文件

    在日常工作中,我们经常需要处理大量的Excel文件,为了保护敏感数据的安全性,我们可能需要对这些文件进行加密,本文主要介绍了如何使用Python实现批量加密Excel文件,需要的可以参考下
    2023-11-11
  • Python实现对比不同字体中的同一字符的显示效果

    Python实现对比不同字体中的同一字符的显示效果

    这篇文章主要介绍了Python实现对比不同字体中的同一字符的显示效果,也就是对比不同字体中某个字的显示效果,这在做设计时非常有用,需要的朋友可以参考下
    2015-04-04
  • pandas读取HTML和JSON数据的实现示例

    pandas读取HTML和JSON数据的实现示例

    Pandas可以直接读取html和JSON数据,本文就来介绍一下pandas读取HTML和JSON数据的实现示例,文中通过示例代码介绍的非常详细,感兴趣的可以了解一下
    2024-01-01
  • python 查找字符串是否存在实例详解

    python 查找字符串是否存在实例详解

    这篇文章主要介绍了python 查找字符串是否存在实例详解的相关资料,需要的朋友可以参考下
    2017-01-01
  • Python通过两个dataframe用for循环求笛卡尔积

    Python通过两个dataframe用for循环求笛卡尔积

    这篇文章主要介绍了Python通过两个dataframe用for循环求笛卡尔积,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-04-04
  • python读取Kafka实例

    python读取Kafka实例

    今天小编就为大家分享一篇python读取Kafka实例,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2019-12-12
  • Python利用字典将两个通讯录文本合并为一个文本实例

    Python利用字典将两个通讯录文本合并为一个文本实例

    这篇文章主要介绍了Python利用字典将两个通讯录文本合并为一个文本实例,具有一定借鉴价值,需要的朋友可以参考下
    2018-01-01
  • 详解Flask框架中Flask-Login模块的使用

    详解Flask框架中Flask-Login模块的使用

    Flask-Login 是一个 Flask 模块,可以为 Flask 应用程序提供用户登录功能。这篇文章将通过一些示例为大家介绍一下Flask-Login模块的使用,需要的可以参考一下
    2023-01-01
  • OpenCV 光流Optical Flow示例

    OpenCV 光流Optical Flow示例

    这篇文章主要为大家介绍了OpenCV 光流Optical Flow示例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2023-04-04
  • 分享python中matplotlib指定绘图颜色的八种方式

    分享python中matplotlib指定绘图颜色的八种方式

    这篇文章主要给大家分享的是python中matplotlib指定绘图颜色的八种方式,在使用matplotlib的pyplot库进行绘图时,经常会发现各种开源代码指定“color”的方式并不一致,下面就向大家展示8种指定color的方式,需要的朋友可以参考一下
    2022-03-03

最新评论