FastAPI中正确使用ApScheduler 解决定时任务运行实例数超限报错

问题原因 #

• 你的check_list_len任务函数内部写了while True死循环，第一次被定时任务触发后就会一直运行，永远不会结束
• APScheduler默认单个任务的最大并发运行实例数为1，后续每5秒触发的新任务实例都会因为前一个实例还在运行，被直接跳过，就会出现你看到的报错
• 你已经通过cron规则设置了每5秒触发一次任务，完全不需要在函数内部再写死循环+sleep逻辑，二者冲突了

修复方案 #

1. 移除check_list_len函数内部的while True和time.sleep(5)逻辑，仅保留单次检查的代码即可，定时触发由APScheduler的cron规则负责
2. 如果确实有需要让同一个任务同时运行多个实例，可以在添加任务时指定max_instances参数，例如：

scheduler.add_job(check_list_len, 'cron', second='*/5', max_instances=2)

（注：此方案不适合你的当前场景，仅作为通用拓展说明）

修正后的完整代码 #

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from loguru import logger
from apscheduler.schedulers.background import BackgroundScheduler

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

test_list = ["1"]*10

# 移除死循环和sleep，仅保留单次检查逻辑
def check_list_len():
    global test_list
    logger.info(f"check_list_len：{len(test_list)}")

@app.on_event('startup')
def init_data():
    scheduler = BackgroundScheduler()
    scheduler.add_job(check_list_len, 'cron', second='*/5')
    scheduler.start()

@app.get("/pop")
async def list_pop():
    global test_list
    test_list.pop(1)
    logger.info(f"current_list_len:{len(test_list)}")


if __name__ == '__main__':
    uvicorn.run(app="main3:app", host="0.0.0.0", port=80, reload=False, debug=False)

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