FastAPI中如何正确使用ApScheduler？

Question

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
import time
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

def check_list_len():
    global test_list
    while True:
        time.sleep(5)
        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)

上面是我的代码，我想通过get请求取出一个元素列表，并设置一个周期性任务不断检查列表中的元素数量，但是当我运行时，总是出现以下错误：

Execution of job "check_list_len (trigger: cron[second='*/5'], next run at: 2021-11-25 09:48:50 CST)" skipped: maximum number of running instances reached (1)
2021-11-25 09:48:50.016 | INFO     | main3:check_list_len:23 - check_list_len：10
Execution of job "check_list_len (trigger: cron[second='*/5'], next run at: 2021-11-25 09:48:55 CST)" skipped: maximum number of running instances reached (1)
2021-11-25 09:48:55.018 | INFO     | main3:check_list_len:23 - check_list_len：10
INFO:     127.0.0.1:55961 - "GET /pop HTTP/1.1" 200 OK
2021-11-25 09:48:57.098 | INFO     | main3:list_pop:35 - current_list_len:9
Execution of job "check_list_len (trigger: cron[second='*/5'], next run at: 2021-11-25 09:49:00 CST)" skipped: maximum number of running instances reached (1)
2021-11-25 09:49:00.022 | INFO     | main3:check_list_len:23 - check_list_len：9

看起来我启动了两项计划任务，只有一项成功，但我只启动了一项任务。我该如何避免这种情况

Answer 1

您正在得到您所要求的行为。您已将 apscheduler 配置为每五秒运行一次

check_list_len

，但您还使该函数运行而不终止 - 只是在无限循环中休眠五秒。该函数永远不会终止，因此

apscheduler

不会再次运行它 - 因为它还没有完成。

使用

apscheduler

时删除实用程序函数内的无限循环 - 它会为您每五秒调用一次该函数：

def check_list_len():
    global test_list  # you really don't need this either, since you're not reassigning the variable
    logger.info(f"check_list_len：{len(test_list)}")

Answer 2

遇到这个包：https://pypi.org/project/fastapi-utilities/

它有两个函数

repeat_at

，其工作方式类似于 cron 和

repeat_every

，它在每隔给定的秒后执行一项任务。

安装它：

pip install fastapi-utilities

你可以这样做：

from fastapi_utilities import repeat_at

@router.on_event("startup")
@repeat_at(cron="*/2 * * * *") #every 2nd minute
async def hey():
    print("hey")