Python使用signal定時結束AsyncIOScheduler任務的問題

Posted on by

在使用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任務內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：