Apscheduler结合flask进行动态任务管理

1. 简介与安装

Apscheduler是一个轻量级的 Python 定时任务调度框架。APScheduler 支持三种调度任务：固定时间间隔，固定时间点（日期），Linux 下的 Crontab 命令。同时，它还支持异步执行、后台执行调度任务。

pip install apscheduler

2. APScheduler基础组成

2.1 调度器(scheduler)

BlockingScheduler: 调度器在当前进程的主线程中运行，会阻塞当前线程。
BackgroundScheduler: 调度器在后台线程中运行，不会阻塞当前线程。
AsyncIOScheduler: 结合asyncio模块一起使用。
GeventScheduler: 程序中使用gevent作为IO模型和GeventExecutor配合使用。
TornadoScheduler: 程序中使用Tornado的IO模型，用 ioloop.add_timeout 完成定时唤醒。
TwistedScheduler: 配合TwistedExecutor，用reactor.callLater完成定时唤醒。
QtScheduler: 应用是一个Qt应用，需使用QTimer完成定时唤醒。

2.2 触发器(trigger)

date是最基本的一种调度，作业任务只会执行一次。参数详见
interval触发器，固定时间间隔触发。参数详见
cron 触发器，在特定时间周期性地触发，和Linux crontab格式兼容。它是功能最强大的触发器。参数详见

2.3 执行器(executor)

执行器是执行调度任务的模块。最常用的 executor 有两种：ProcessPoolExecutor 和 ThreadPoolExecutor

2.4 任务存储(job store)

存储预定的作业。默认job store只是将作业保存在内存中，但其他job store将它们存储在各种数据库中。作业的数据在保存到持久job store时被序列化，并在从它加载回来时被反序列化。job store（默认存储除外）不会将作业数据保存在内存中，而是充当中间人，用于在后端保存、加载、更新和搜索作业。job store绝不能在调度程序之间共享。

3. Flask框架实践

3.1 定时任务初始化

apscheduler在app创建时进行初始化, scheduler在初始化器类中进行创建,初始化器类采用单例设计, 在整个flask实例上下文中都是同一个对象,以满足通过接口进行动态更新任务的需求

# src.initializer.py
from apscheduler.schedulers.background import BackgroundSchedulerdef singleton(cls):_instance = {}def _singleton(*args, **kargs):if cls not in _instance:_instance[cls] = cls(*args, **kargs)return _instance[cls]return _singleton@singleton
class Initializer:def __init__(self):self.scheduler = BackgroundScheduler()

scheduler初始化之后可以通过add_job添加一些固定的定时任务, 再通过start方法就可以正式启动定时任务

# main.py
from flask_apscheduler import APScheduler
from src.initializer import Initializerdef create_app():app = Flask(__name__)app.config.from_object(Config)initializer = Initializer()update_job = {"id": "print_jobs","func": "src.utils.common.scheduler:update_jobs","args": (initializer.scheduler,),"trigger": "interval","seconds": 10,}initializer.scheduler.add_job(**update_job)initializer.scheduler.start()return app

3.2 通过接口进行任务动态增减

在初始化flask实例时进行了apscheduler单例化, 可以动态的通过flask接口进行定时任务的删除和新增.使用唯一的job_id进行任务的获取, 删除和新增.

from flask import current_app@app.route("/test_apscheduler")
def test_apscheduler():"""测试定时任务动态进行删除和新增任务"""ping_baidu = {"id": "ping_baidu","func": "src.utils.common.scheduler:ping","args": ("baidu.com",),"trigger": "interval","seconds": 10,}job = current_app.apscheduler.get_job(job_id="ping")if job:current_app.apscheduler.remove_job(job_id="ping")current_app.apscheduler.add_job(**ping_baidu)return http_json({"msg": "ok"})

3.3 通过定时任务自我进行任务动态管理

如上在进行flask初始化时有预设一个update_job的定时任务,传参为scheduler调度器, 每2分钟运行一次, 在内部实现任务从配置文件或数据库读取,然后进行任务的删除、更新、新增操作，以此达到定时任务的自我管理。

# 伪代码src.utils.common.scheduler.pydef update_job(scheduler):"""定时任务自我管理,动态更新job"""jobs = scheduler.get_jobs()new_jobs, expired_jobs = get_job_from_db(jobs)for job in expired_jobs:scheduler.remove_job(job_id=job.id)for job in new_jobs:scheduler.add_job(job)

4. 踩坑点

4.1 多进程部署,定时任务重复启动

解决思路: 在启动定时任务时, 设置文件锁, 当不能获取到文件锁时, 不再启动任务


def create_app():app =Flask(__name__)# 启动定时任务scheduler_init(app)return appdef scheduler_init(app):"""保证系统只启动一次定时任务"""if platform.system() != 'Windows':fcntl = __import__("fcntl")f = open('scheduler.lock', 'wb')try:fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)scheduler.init_app(app)scheduler.start()app.logger.debug('Scheduler Started,---------------')except:passdef unlock():fcntl.flock(f, fcntl.LOCK_UN)f.close()atexit.register(unlock)else:msvcrt = __import__('msvcrt')f = open('scheduler.lock', 'wb')try:msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1)scheduler.init_app(app)scheduler.start()app.logger.debug('Scheduler Started,----------------')except:passdef _unlock_file():try:f.seek(0)msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)except:passatexit.register(_unlock_file)

4.2 多主机分布式部署,定时任务重复启动

apscheduler本身不支持分布式,因此分布式部署还是会出现重复启动问题.

可以多apscheduler进行分布式重写解决, 重写思路: 分布式场景下使用APScheduler 、apscheduler分布式调度