如何在Python中使用异步模式实现定时任务调度系统？-Python教程-PHP中文网

asyncio与APScheduler混用易触发RuntimeError: Event loop is closed，因BackgroundScheduler启独立线程和循环，不兼容asyncio单loop约束；应改用AsyncIOScheduler（job须为async def）、aiocron或手写create_task+sleep方案，并确保HTTP/DB操作全程异步。

如何在python中使用异步模式实现定时任务调度系统？

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，并确保它真正在后台持续运行：

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())
</p>

注意点：

crontab 字符串末尾多一个 * 表示秒级（*/5 * * * * *），标准 cron 是 5 段，aiocron 支持 6 段（加秒）
装饰的函数必须是 async def，不能是普通函数
没有 .start() 或 .shutdown() —— 它随 loop 启动/关闭自动生效

AsyncIOScheduler 要求 job 函数全部是协程，且不能混用 time.sleep

如果你已有 APScheduler 使用经验，想迁移到 async 场景，必须用 AsyncIOScheduler 并严格遵守协程约束。

错误示范（会阻塞整个 event loop）：

Python Dependency Candidate Fetcher 2

根据包的能力以及 Jupyter 环境中的导入语句，从候选列表中挑选最适合用户表格工作流的 Python 包。

下载

from apscheduler.schedulers.asyncio import AsyncIOScheduler
import time
<p>def bad_job():  # ❌ 普通函数 + time.sleep → 同步阻塞
time.sleep(2)
print("done")
</p>

正确写法：

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()
</p>

关键细节：

不能在 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-py v4.2+ 的 redis.Redis(..., decode_responses=True, protocol=3) + await 调用