asyncio与APScheduler混用易触发RuntimeError: Event loop is closed，因BackgroundScheduler启独立线程和循环，不兼容asyncio单loop约束；应改用AsyncIOScheduler（job须为async def）、aiocron或手写create_task+sleep方案，并确保HTTP/DB操作全程异步。
asyncio + apscheduler 会出 RuntimeError: Event loop is closed
直接在异步环境中混用 APScheduler（尤其是 BackgroundScheduler）大概率触发这个错误。因为 APScheduler 默认启动自己的线程和独立事件循环，而 asyncio 要求协程必须运行在同一个活跃的 asyncio.get_event_loop() 中。
它不感知你的主协程生命周期，关服务时容易残留线程、关闭 loop 后又试图调度任务。
解决思路不是“怎么让它跑起来”，而是“换掉它”——改用原生支持 async 的调度器：
-
asyncio.create_task()+asyncio.sleep()手写简单轮询（适合固定间隔、少量任务） -
aiocron：轻量、纯 async，语法接近 cron 表达式，但不支持持久化或集群 -
apscheduler[asyncio]的AsyncIOScheduler：需显式传入运行中的 loop，且所有 job 函数必须是async def
用 aiocron 定义每 5 秒执行一次的异步任务
aiocron 是最顺手的入门选择，安装后无需管理 loop 生命周期，自动绑定到当前运行的 event loop。
pip install aiocron
示例：定义一个带参数的异步 job，并确保它真正在后台持续运行：
立即学习“Python免费学习笔记（深入）”；
import aiocron
import asyncio
<p>@aiocron.crontab('<em>/5 </em> <em> </em> <em> </em>') # 每 5 秒
async def heartbeat():
print("ping at", asyncio.get_event_loop().time())</p><h1>必须显式启动 —— 不调用 run_forever()，程序会立即退出</h1><p>async def main():
await asyncio.Event().wait() # 阻塞住，保持事件循环运行</p><p>asyncio.run(main())
注意点：
- crontab 字符串末尾多一个
*表示秒级（*/5 * * * * *），标准 cron 是 5 段，aiocron支持 6 段（加秒） - 装饰的函数必须是
async def，不能是普通函数 - 没有
.start()或.shutdown()—— 它随 loop 启动/关闭自动生效
AsyncIOScheduler 要求 job 函数全部是协程，且不能混用 time.sleep
如果你已有 APScheduler 使用经验，想迁移到 async 场景，必须用 AsyncIOScheduler 并严格遵守协程约束。
错误示范（会阻塞整个 event loop）：
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import time
<p>def bad_job(): # ❌ 普通函数 + time.sleep → 同步阻塞
time.sleep(2)
print("done")
正确写法：
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler
<p>async def good_job():
await asyncio.sleep(2) # ✅ 协程挂起，不阻塞
print("done")</p><p>scheduler = AsyncIOScheduler()
scheduler.add_job(good_job, 'interval', seconds=5)
scheduler.start()</p><h1>注意：必须让 loop 持续运行，否则 scheduler 无机会触发</h1><p>asyncio.get_event_loop().run_forever()
关键细节：
- 不能在 job 里调用
requests.get()，得换成aiohttp.ClientSession - 数据库操作必须用 async 驱动（如
asyncpg、aiomysql），不能用psycopg2 -
scheduler.shutdown()是异步的，要await scheduler.shutdown()
定时任务中访问数据库或 HTTP 接口必须全程 async
很多人卡在“任务能跑，但一发请求就报错”。根本原因是：async 任务里调了同步库，导致 event loop 被独占，后续所有协程冻结。
典型错误链：
- 用
requests发 HTTP 请求 → 报RuntimeError: Cannot enter into task或 CPU 100% - 用
sqlite3.connect()→ 磁盘 IO 阻塞 loop，其他定时任务延迟甚至丢失 - 用
redis.Redis()（非 async 版）→ 连接池阻塞，超时堆积
对应替换方案：
- HTTP：用
aiohttp或httpx.AsyncClient - PostgreSQL：用
asyncpg（推荐）或aiopg - Redis：用
aioredis（v2）或redis-pyv4.2+ 的redis.Redis(..., decode_responses=True, protocol=3)+await调用
哪怕只有一处用了 time.sleep(1) 或 print() 太频繁，都可能挤压 loop 调度窗口——异步定时任务的稳定性，取决于最慢、最“同步”的那个环节。
