首页 > 其他资讯 > FastAPI + APScheduler 进阶：任务持久化 + 分布式锁

FastAPI + APScheduler 进阶：任务持久化 + 分布式锁

时间：26-04-23

为什么需要持久化？

今天我们来深入聊聊APScheduler的两个进阶配置：任务持久化和分布式锁。这两个配置，可以说是让你的定时任务从“能跑”的玩具，升级为“生产可用”的可靠工具的关键一步。

免费影视、动漫、音乐、游戏、小说资源长期稳定更新！ 👉 点此立即查看 👈

你是否有过这样的经历？用APScheduler写了个定时任务，跑得好好的，结果服务一重启，所有任务都消失了。或者，为了提高可用性，部署了多个服务实例，却发现同一个定时任务被重复执行了三遍？别担心，这些坑，很多开发者都踩过。接下来，我们就来一一填平它们。

默认情况下，APScheduler为了追求极致的轻量和速度，会将所有任务信息存储在内存里。这带来了一个显而易见的问题：一旦服务进程终止或重启，内存中的数据就全部丢失了，那些精心配置的定时任务自然也无影无踪。

from apscheduler.schedulers.asyncio import AsyncIOScheduler
scheduler = AsyncIOScheduler()
scheduler.add_job(my_task, 'interval', seconds=60)
scheduler.start()

上面这段代码运行起来没问题，但它的状态是“脆弱”的。想象一下，如果你配置了一个每天凌晨3点执行的关键数据同步任务，你还敢轻易重启服务吗？显然，我们需要一个更可靠的方案。

持久化方案：SQLAlchemy JobStore

幸运的是，APScheduler设计之初就考虑到了这一点，它支持将任务状态持久化到多种后端存储中。其中，基于SQLAlchemy的JobStore因其通用性和稳定性，成为最常用的选择。它支持包括MySQL、PostgreSQL、SQLite在内的多种关系型数据库。

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
scheduler = AsyncIOScheduler(jobstores=jobstores)

配置完成后，所有的任务定义、触发时间、执行状态等信息，都会被安全地记录在指定的数据库（例如这里的`jobs.sqlite`文件）中。这样一来，无论服务因何种原因重启，在重新初始化调度器时，它都能从数据库里恢复出之前的所有任务，真正做到“任务不丢”。

多实例部署的灾难

持久化解决了单点重启的问题，但现代应用架构往往追求高可用，多实例部署是常态。这时，一个新的“坑”就出现了。

假设你的应用使用FastAPI，并且用Kubernetes或类似工具部署了3个实例。如果每个实例都独立启动了自己的APScheduler，那么每个实例都会去加载并执行数据库中记录的同一个定时任务。结果就是，原本计划凌晨3点执行一次的数据同步，会被重复执行三次。

# 实例 1、2、3 都会执行这个任务
scheduler.add_job(sync_data, 'cron', hour=3)

这无疑是灾难性的。如果任务是同步数据，可能导致数据混乱；如果是发送邮件或推送消息，用户就会收到多份重复通知，体验极差。因此，仅仅有持久化还不够，我们还需要一种机制来保证任务在分布式环境下的“唯一执行”。

分布式锁：避免重复执行

APScheduler本身并未内置分布式锁功能，但这并不妨碍我们利用现有的工具来实现它。一个广泛采用的方案是借助Redis来实现一个轻量级的分布式锁。

核心思路非常直观：在任务开始执行前，所有实例都尝试去获取一个唯一的“锁”；只有一个实例能成功抢到这把锁，只有抢到锁的实例才执行实际的任务逻辑，其他实例则自动放弃执行。

import redis
from contextlib import contextmanager

redis_client = redis.Redis(host='localhost', port=6379, db=0)

@contextmanager
def distributed_lock(lock_key: str, expire: int = 60):
    """分布式锁上下文管理器"""
    acquired = redis_client.set(lock_key, '1', nx=True, ex=expire)
    try:
        yield acquired
    finally:
        if acquired:
            redis_client.delete(lock_key)

在实际使用时，我们只需要用这个锁包裹住任务的核心逻辑即可：

def sync_data():
    with distributed_lock('sync_data_lock') as acquired:
        if not acquired:
            print('其他实例正在执行，跳过')
            return

        # 执行同步逻辑
        print('开始同步数据...')

通过这样的设计，即使有3个、甚至30个实例同时触发任务，也只有一个实例能成功获取名为`sync_data_lock`的锁并执行任务，其他实例都会安静地跳过。这就完美解决了多实例重复执行的问题。

完整示例：FastAPI 集成

将持久化存储和分布式锁组合起来，我们就能构建出一个健壮、生产可用的定时任务系统。下面是一个与FastAPI框架集成的完整示例：

from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import redis

app = FastAPI()

# 1. 持久化配置
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
scheduler = AsyncIOScheduler(jobstores=jobstores)

# 2. Redis 客户端
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def distributed_lock(lock_key: str, expire: int = 300):
    """分布式锁装饰器"""
    def decorator(func):
        async def wrapper(*args, **kwargs):
            acquired = redis_client.set(lock_key, '1', nx=True, ex=expire)
            if not acquired:
                return None
            try:
                return await func(*args, **kwargs)
            finally:
                redis_client.delete(lock_key)
        return wrapper
    return decorator

@distributed_lock('daily_sync_lock')
async def daily_sync():
    """每日数据同步任务"""
    print('执行数据同步...')

@app.on_event('startup')
async def startup():
    scheduler.add_job(
        daily_sync,
        'cron',
        hour=3,
        id='daily_sync',
        replace_existing=True
    )
    scheduler.start()

@app.on_event('shutdown')
async def shutdown():
    scheduler.shutdown()

这个方案清晰地展示了如何将两大核心配置融入一个真实的Web服务中，确保了定时任务的高可靠性与唯一性。