本文详解如何将 aiohttp 的 ClientSession 与 APScheduler 的异步调度器协同使用，避免会话提前关闭或事件循环冲突，确保多个定时 HTTP 请求复用同一连接池并稳定运行。
本文详解如何将 aiohttp 的 `clientsession` 与 apscheduler 的异步调度器协同使用，避免会话提前关闭或事件循环冲突，确保多个定时 http 请求复用同一连接池并稳定运行。
在异步 Python 应用中，将 aiohttp.ClientSession 与 APScheduler（尤其是 AsyncIOScheduler）结合时，一个常见误区是试图在普通函数中“启动”调度器并传入会话——这会导致 ClientSession 生命周期管理失控：若会话在 jobs() 函数作用域内创建并退出 async with 块，其底层连接池将被关闭，而后续调度任务仍尝试使用已关闭的 session，引发 RuntimeError: Session is closed。
正确的做法是将 ClientSession 的生命周期提升至主协程顶层，并确保其存活时间覆盖整个调度周期。以下是推荐实现：
import asyncio
import aiohttp
from apscheduler.schedulers.asyncio import AsyncIOScheduler
async def fetch(session, url, timeout=3):
try:
async with session.get(url, ssl=False, timeout=timeout) as response:
return await response.text(), response.status
except Exception as e:
print(f"Failed to fetch {url}: {e}")
return None, 0
async def GOESX_job(session):
url = 'https://services.swpc.noaa.gov/json/goes/primary/xrays-6-hour.json'
response, status = await fetch(session, url)
if response:
print(f'GOESX — Status: {status}, Length: {len(response)} chars')
async def GOESp_job(session):
url = 'https://services.swpc.noaa.gov/json/goes/primary/integral-protons-6-hour.json'
response, status = await fetch(session, url)
if response:
print(f'GOESp — Status: {status}, Length: {len(response)} chars')
async def main():
scheduler = AsyncIOScheduler()
# ✅ 正确：ClientSession 在主协程中创建，与 scheduler 同生命周期
async with aiohttp.ClientSession() as session:
# 注册任务时直接传入活跃 session 实例
scheduler.add_job(GOESX_job, 'interval', seconds=5, args=[session])
scheduler.add_job(GOESp_job, 'interval', seconds=10, args=[session])
scheduler.start()
# ⚠️ 关键：保持事件循环运行，且不退出 session 作用域
# 使用 await asyncio.sleep(float("inf")) 阻塞主协程，而非 run_forever()
try:
await asyncio.sleep(float("inf"))
except KeyboardInterrupt:
print("Shutting down...")
finally:
scheduler.shutdown(wait=True)
# 启动入口（Python 3.7+ 推荐方式）
if __name__ == "__main__":
asyncio.run(main())
注意事项与最佳实践：
- 不要混用 asyncio.get_event_loop().run_forever()：它会接管事件循环控制权，与 asyncio.run() 冲突，且无法优雅关闭 session 和 scheduler。
- 避免在 job 函数内新建 session：违背复用连接池初衷，增加开销与资源泄漏风险。
- 异常处理不可省略：网络请求易失败，务必在 fetch() 中捕获 aiohttp.ClientError、asyncio.TimeoutError 等，防止单个失败任务中断整个调度器。
- SSL 配置需谨慎：示例中 ssl=False 仅用于测试；生产环境应启用证书验证（移除该参数或显式设为 True），必要时配置 trust_env=True 或自定义 connector。
- 调度器关闭：scheduler.shutdown(wait=True) 确保所有正在执行的任务完成后再退出，配合 async with 可保证 session 清理安全。
通过上述结构，你既能灵活在各 job 中动态生成 URL（如根据时间戳、外部状态计算），又能高效复用连接、精准控制调度节奏，为构建健壮的异步数据采集服务打下坚实基础。
