pip install apscheduler

# 項(xiàng)目結(jié)構(gòu)建議
# app/
#   __init__.py
#   main.py       # FastAPI 應(yīng)用創(chuàng)建和事件處理
#   scheduler.py  # 調(diào)度器配置和任務(wù)定義
#   tasks.py      # 具體的任務(wù)函數(shù)

# app/scheduler.py
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor

# 1. 配置組件
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')  # 使用SQLite持久化
}
executors = {
    'default': ThreadPoolExecutor(20)  # 線程池執(zhí)行
}
job_defaults = {
    'coalesce': False,  # 錯(cuò)過(guò)的任務(wù)是否合并執(zhí)行（一般False）
    'max_instances': 3  # 同一個(gè)任務(wù)同時(shí)運(yùn)行的最大實(shí)例數(shù)
}

# 2. 創(chuàng)建調(diào)度器實(shí)例
scheduler = AsyncIOScheduler(
    jobstores=jobstores,
    executors=executors,
    job_defaults=job_defaults,
    timezone="Asia/Shanghai"  # 時(shí)區(qū)！時(shí)區(qū)！時(shí)區(qū)！重要的事說(shuō)三遍
)

# 3. 定義任務(wù)函數(shù) (可以放在同文件，也可從其他模塊導(dǎo)入)
def my_sync_job():
    print("同步任務(wù)執(zhí)行了！")

async def my_async_job():
    print("異步任務(wù)執(zhí)行了！")
    # 這里可以愉快地調(diào)用其他async函數(shù)

# 4. 添加任務(wù)的函數(shù) (通常在應(yīng)用啟動(dòng)時(shí)調(diào)用)
def add_jobs():
    # 間隔任務(wù)：每30秒執(zhí)行一次
    scheduler.add_job(my_sync_job, 'interval', seconds=30, id='sync_interval_job')
    # Cron任務(wù)：每分鐘的第30秒執(zhí)行
    scheduler.add_job(my_async_job, 'cron', second=30, id='async_cron_job')
    # 單次任務(wù)：2023年10月1日?qǐng)?zhí)行
    # scheduler.add_job(xxx, 'date', run_date='2023-10-01 00:00:00')    

# app/main.py
from fastapi import FastAPI
from .scheduler import scheduler, add_jobs

app = FastAPI(title="定時(shí)任務(wù)演示")

@app.on_event("startup")
async def startup_event():
    # 應(yīng)用啟動(dòng)時(shí)，添加任務(wù)并啟動(dòng)調(diào)度器
    if not scheduler.running:
        add_jobs()
        scheduler.start()
        print("APScheduler 已啟動(dòng)")

@app.on_event("shutdown")
async def shutdown_event():
    # 應(yīng)用關(guān)閉時(shí)，優(yōu)雅地關(guān)閉調(diào)度器
    if scheduler.running:
        scheduler.shutdown()
        print("APScheduler 已關(guān)閉")

#from contextlib import asynccontextmanager
#@asynccontextmanager
#async def lifespan(app: FastAPI):
#    # 應(yīng)用啟動(dòng)時(shí)，添加任務(wù)并啟動(dòng)調(diào)度器
#    if not scheduler.running:
#        add_jobs()
#        scheduler.start()
#        print("APScheduler 已啟動(dòng)")
#    yield
#    # 應(yīng)用關(guān)閉時(shí)，優(yōu)雅地關(guān)閉調(diào)度器
#    if scheduler.running:
#        scheduler.shutdown()
#        print("APScheduler 已關(guān)閉")
#app = FastAPI(title="定時(shí)任務(wù)演示", lifespan=lifespan)

@app.get("/")
async def root():
    return {"message": "Hello World"}

# 可選：提供一個(gè)API來(lái)手動(dòng)觸發(fā)或查看任務(wù)狀態(tài)
@app.get("/jobs")
async def list_jobs():
    jobs = scheduler.get_jobs()
    return {"jobs": [{"id": j.id, "next_run": str(j.next_run_time)} for j in jobs]}    

# 在 scheduler.py 或 startup 事件中
import os
import sys

def try_acquire_lock(lock_file):
    try:
        import fcntl
        f = open(lock_file, 'w')
        # 嘗試獲取非阻塞的獨(dú)占鎖
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        return f  # 返回文件對(duì)象，保持打開(kāi)狀態(tài)以持有鎖
    except (BlockingIOError, ImportError):
        # 獲取失?。ㄆ渌M(jìn)程已持有鎖）或不支持的系統(tǒng)
        return None

lock_file = "/tmp/fastapi_scheduler.lock"
lock_fd = try_acquire_lock(lock_file)

@app.on_event("startup")
async def startup_event():
    if lock_fd is not None:
        # 只有拿到鎖的進(jìn)程才啟動(dòng)調(diào)度器
        if not scheduler.running:
            add_jobs()
            scheduler.start()
            print(f"進(jìn)程 {os.getpid()} 成功啟動(dòng) APScheduler")
    else:
        print(f"進(jìn)程 {os.getpid()} 未獲得鎖，跳過(guò)調(diào)度器啟動(dòng)")        

# 假設(shè)使用SQLAlchemy ORM
from sqlalchemy.ext.asyncio import AsyncSession
from your_app.models import SchedulerLock
import datetime

async def acquire_db_lock(session: AsyncSession, timeout_minutes=10):
    try:
        # 嘗試插入一條鎖記錄，host和pid標(biāo)識(shí)當(dāng)前進(jìn)程
        lock = SchedulerLock(
            id=1,  # 固定ID
            host="my_host",
            pid=os.getpid(),
            last_heartbeat=datetime.datetime.utcnow()
        )
        session.add(lock)
        await session.commit()
        return True  # 插入成功，獲得鎖
    except IntegrityError:  # 唯一約束沖突，記錄已存在
        await session.rollback()
        # 檢查已有的鎖是否已過(guò)期
        existing_lock = await session.get(SchedulerLock, 1)
        if existing_lock and (datetime.datetime.utcnow() - existing_lock.last_heartbeat).seconds > timeout_minutes * 60:
            # 鎖已過(guò)期，更新為當(dāng)前進(jìn)程
            existing_lock.host = "my_host"
            existing_lock.pid = os.getpid()
            existing_lock.last_heartbeat = datetime.datetime.utcnow()
            await session.commit()
            return True
        return False  # 未能獲得鎖

# 在 startup 事件中調(diào)用 acquire_db_lock 判斷