最近，想要实现一个功能，就是添加定时任务的同时记录其任务信息到数据库中，可以通过接口查看设定了哪些定时任务，经过调研，发现大家公认的Python最好用的定时任务框架是APScheduler。

Python Scheduler（APScheduler）是一个Python库，它允许您安排稍后要执行的Python代码，只需一次或定期执行。 您可以在您的同时添加新的作业或删除旧的旧作业。 如果在数据库中存储工作，他们还将存活调度程序重新启动并保持状态。 调度程序重新启动时，它将运行它在脱机时应该运行的所有作业

在网上找了一堆资料，包括官方教程，但对于和MySQL协作使用的方面语焉不详，因此，这里做个记录。

本文参考了官方教程以及大佬1的博客、大佬2的博客

项目使用的环境如下：

• Python 3.7
• MySQL
• APScheduler
• Navicat for MySQL

APScheduler Startup

首先，安装APScheduler：

pip install apscheduler

然后就可以编写简单的demo了：

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def job():
    # 输出当前时间
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

# BlockingScheduler
scheduler = BlockingScheduler()
scheduler.add_job(job, 'cron', day_of_week='0-4', hour=6, minute=30)

这个demo表示每周的星期一到星期五的六点半报时

APScheduler基础教程

具体见大佬2的博客，这里不做赘述

这里总结一下常用的任务模式。

interval间隔时间任务

即每隔一段时间执行一次，可能的字段为：

例子：

#表示每隔3秒执行一次任务
scheduler.add_job(tok, 'interval', seconds=3)

#表示每隔3天17时19分07秒执行一次任务
sched.add_job(my_job, 'interval',days  = 03,hours = 17,minutes = 19,seconds = 07)

date定点任务

即在某个时刻运行，但仅执行一次，可能的字段为：

栗子：

# 任务将在2009年11月6日00:00:00执行
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])

# 任务将在2009年11月6日16:30:05执行
sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text'])

cron某一时刻任务

可能的字段为：

其中，int/str 表示参数既可以是int类型，也可以是str类型；datetime/str表示参数既可以是datetime类型，也可以是str类型

例子：

#表示2017年3月22日17时19分07秒执行该程序
sched.add_job(my_job, 'cron', year=2017,month = 03,day = 22,hour = 17,minute = 19,second = 07)

#表示任务在6,7,8,11,12月份的第三个星期五的00:00,01:00,02:00,03:00 执行该程序
sched.add_job(my_job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')

#表示从星期一到星期五5:30（AM）直到2014-05-30 00:00:00
sched.add_job(my_job(), 'cron', day_of_week='mon-fri', hour=5, minute=30,end_date='2014-05-30')

#表示每5秒执行该程序一次，相当于interval 间隔调度中seconds = 5
sched.add_job(my_job, 'cron',second = '*/5') 作者：燕洼仙草 https://www.bilibili.com/read/cv7868920/ 出处：bilibili

APScheduler + MySQL

不过我们肯定并不满足于实现一个小小的demo。

如果我们想要连接MySQL实现持久化存储应该怎么做呢？

这里需要配置调度器：

scheduler = BackgroundScheduler({
    'apscheduler.jobstores.default': {
        'type': 'sqlalchemy',
        'url': 'mysql+pymysql://username:password@127.0.0.1:3306/dbname?charset=utf8',
        'tablename': 'api_job'
    },
    'apscheduler.executors.default': {
        'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
        'max_workers': '20'
    },
    'apscheduler.executors.processpool': {
        'type': 'processpool',
        'max_workers': '10'
    },
    'apscheduler.job_defaults.coalesce': 'false',
    'apscheduler.job_defaults.max_instances': '10',
    'apscheduler.timezone': 'UTC',
})

其中，apscheduler.jobstores.default字段表示存储器的配置，字段说明如下：

• type表示数据库的类型，MySQL属于SQLAlchemy，因此这里值填写sqlalchemy
• url表示存储器访问呢的地址，其中的这些字段需要配置：
  • username为你的数据库的用户名
  • password为对应的账号的密码
  • dbname为数据库名称
• tablename表示将定时任务数据存储的表名

然后就可以使用配置好的scheduler添加任务了：

scheduler.add_job(tok, 'interval', seconds=3, id=job_id, coalesce=True, replace_existing=True)

其中id表示任务的唯一标识符，coalesce表示忽略服务器宕机时间段内的任务执行（否则就会出现服务器恢复之后一下子执行多次任务的情况），replace_existing表示如果有重名的任务，直接覆盖

添加任务后就可以看到数据库多出了内容：

添加任务后的数据库

这里使用了中文字符串作为id，可以语义地表示任务

然鹅，要想任务存储其他信息怎么办，这里思来想去，决定在数据库创建一个新的表，用于记录任务的详细信息，方便后续查看，每次添加任务时往该表插值即可，删除任务时也删除对应的数据，由此实现任务数据的同步。

不过需要注意的是，在添加重复名称的定时任务时，设置replace_existing=True，直接覆盖已有的任务；相应地，在对其任务信息表，我们采用REPLACE替代INSERT语句来执行插入操作，其使用方法和INSERT完全一样，但如果表中有重复的数据，则会直接覆盖，其优势不言而喻。由此，实现了协同操作。

其数字孪生任务信息表的字段可为：

定时任务详情表