Python使用signal定时结束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()  # 永远运行
© 版权声明
THE END
喜欢就支持一下吧
点赞0 分享
评论 抢沙发

请登录后发表评论

    暂无评论内容