如何将aiohttp与APScheduler结合使用？缺失代码问题咨询

核心问题分析 #

你的代码存在两个关键问题：

1. 异步函数jobs(scheduler)不能直接调用，必须将其提交到事件循环中执行
2. 原代码中async with aiohttp.ClientSession()会在jobs协程执行完毕后立刻关闭Session，导致后续定时任务调用已关闭的Session报错

修正方案 #

将???处替换为asyncio.ensure_future(jobs(scheduler))（或asyncio.get_event_loop().create_task(jobs(scheduler))），同时调整Session的生命周期管理逻辑，避免提前关闭。完整修正后的代码如下：

from apscheduler.schedulers.asyncio import AsyncIOScheduler
import asyncio
import aiohttp

async def fetch(session, url, timeout=3):
    async with session.get(url, ssl=False, timeout=timeout) as response:
        return await response.text(), response.status

async def GOESX_job(session):
    url = 'https://services.swpc.noaa.gov/json/goes/primary/xrays-6-hour.json'
    try:
        response, status = await fetch(session, url)
        print('GOESX', status)
    except Exception as e:
        print(f'GOESX任务执行失败: {str(e)}')

async def GOESp_job(session):
    url = 'https://services.swpc.noaa.gov/json/goes/primary/integral-protons-6-hour.json'
    try:
        response, status = await fetch(session, url)
        print('GOESp', status)
    except Exception as e:
        print(f'GOESp任务执行失败: {str(e)}')

async def jobs(scheduler):
    # 手动创建Session，不使用async with，避免提前关闭
    session = aiohttp.ClientSession()
    scheduler.add_job(GOESX_job, 'interval', seconds=5, args=[session])
    scheduler.add_job(GOESp_job, 'interval', seconds=10, args=[session])
    
    # 保持协程运行，直到程序终止
    try:
        await asyncio.Event().wait()
    finally:
        # 程序退出时关闭Session
        await session.close()

scheduler = AsyncIOScheduler()
# 将异步任务提交到事件循环
asyncio.ensure_future(jobs(scheduler))

scheduler.start()
try:
    asyncio.get_event_loop().run_forever()
except (KeyboardInterrupt, SystemExit):
    pass
finally:
    scheduler.shutdown()

关键修改说明 #

• 替换???处代码：通过asyncio.ensure_future将异步的jobs函数提交到事件循环，确保其能被执行
• 调整Session管理：去掉async with，手动创建并持有Session，直到程序退出时才关闭，保证定时任务全程可用同一个Session
• 添加异常捕获：避免单个任务的网络错误或超时导致整个程序崩溃
• 完善退出逻辑：捕获KeyboardInterrupt和SystemExit信号，确保Scheduler和Session都能正确关闭

内容的提问来源于stack exchange，提问作者Serge Stroobandt