定时任务 APScheduler

简单例子

import time
from apscheduler.schedulers.blocking import BlockingScheduler
def func():
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func  time :',ts)
def func2():
    #耗时2S
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func2 time：',ts)
    time.sleep(2)
def dojob():
    #创建调度器：BlockingScheduler
    scheduler = BlockingScheduler()
    #添加任务,时间间隔2S
    scheduler.add_job(func, 'interval', seconds=2, id='test_job1')
    #添加任务,时间间隔5S
    scheduler.add_job(func2, 'interval', seconds=3, id='test_job2')
    scheduler.start()
dojob()

应用定时发送消息的例子

这里以一个整点报时的功能为例，来介绍定时任务的使用。
新建文件 awesome/plugins/scheduler.py，编写代码如下：

from datetime import datetime
import nonebot
import pytz
from aiocqhttp.exceptions import Error as CQHttpError
@nonebot.scheduler.scheduled_job('cron', hour='*')
async def _():
    bot = nonebot.get_bot()
    now = datetime.now(pytz.timezone('Asia/Shanghai'))
    try:
        await bot.send_group_msg(group_id=672076603,
                                 message=f'现在{now.hour}点整啦！')
    except CQHttpError:
        pass

这里最主要的就是第 8 行，nonebot.scheduler.scheduled_job() 是一个装饰器，第一个参数是触发器类型（这里是 cron，表示使用 Cron 类型的触发参数）。这里 hour='*' 表示每小时都执行，minute和 second 不填时默认为 0，也就是说装饰器所装饰的这个函数会在每小时的第一秒被执行。
除了 cron，还有两种触发器类型 interval 和 date。例如，你可以使用 nonebot.scheduler.scheduled_job('interval', minutes=10) 来每十分钟执行一次任务。
限于篇幅，这里无法给出太详细的接口介绍，nonebot.scheduler 是一个 APScheduler 的 AsyncIOScheduler 对象，因此关于它的更多使用方法，可以参考 APScheduler 的官方文档。

1 简介

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

2 安装

使用 pip 包管理工具安装 APScheduler 是最方便快捷的。

pip install APScheduler
# 如果出现因下载失败导致安装不上的情况，建议使用代理
pip --proxy http://代理ip:端口 install APScheduler

3 使用步骤

APScheduler 使用起来还算是比较简单。运行一个调度任务只需要以下三部曲。

新建一个 schedulers (调度器) 。 添加一个调度任务(job stores)。 运行调度任务。

下面是执行每 2 秒报时的简单示例代码：

import datetime
import time
from apscheduler.schedulers.background import BackgroundScheduler
def timedTask():
    print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])
if __name__ == '__main__':
    # 创建后台执行的 schedulers
    scheduler = BackgroundScheduler()  
    # 添加调度任务
    # 调度方法为 timedTask，触发器选择 interval(间隔性)，间隔时长为 2 秒
    scheduler.add_job(timedTask, 'interval', seconds=2)
    # 启动调度任务
    scheduler.start()
    while True:
        print(time.time())
        time.sleep(5)

4 基础组件

APScheduler 有四种组件，分别是：调度器(scheduler)，作业存储(job store)，触发器(trigger)，执行器(executor)。

• schedulers（调度器）
  它是任务调度器，属于控制器角色。它配置作业存储器和执行器可以在调度器中完成，例如添加、修改和移除作业。
  
• triggers（触发器）
  描述调度任务被触发的条件。不过触发器完全是无状态的。
  
• job stores（作业存储器）
  任务持久化仓库，默认保存任务在内存中，也可将任务保存都各种数据库中，任务中的数据序列化后保存到持久化数据库，从数据库加载后又反序列化。
  

• executors（执行器）
  负责处理作业的运行，它们通常通过在作业中提交指定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。
  

  4.1 schedulers（调度器）

  我个人觉得 APScheduler 非常好用的原因。它提供 7 种调度器，能够满足我们各种场景的需要。例如：后台执行某个操作，异步执行操作等。调度器分别是：

• BlockingScheduler : 调度器在当前进程的主线程中运行，也就是会阻塞当前线程。

• BackgroundScheduler : 调度器在后台线程中运行，不会阻塞当前线程。
• AsyncIOScheduler : 结合 asyncio 模块（一个异步框架）一起使用。
• GeventScheduler : 程序中使用 gevent（高性能的Python并发框架）作为IO模型，和 GeventExecutor 配合使用。
• TornadoScheduler : 程序中使用 Tornado（一个web框架）的IO模型，用 ioloop.add_timeout 完成定时唤醒。
• TwistedScheduler : 配合 TwistedExecutor，用 reactor.callLater 完成定时唤醒。
• QtScheduler : 你的应用是一个 Qt 应用，需使用QTimer完成定时唤醒。

  4.2 triggers（触发器）

  APScheduler 有三种内建的 trigger:
  1）date 触发器
  date 是最基本的一种调度，作业任务只会执行一次。它表示特定的时间点触发。它的参数如下：

date 触发器使用示例如下：

from datetime import datetime
from datetime import date
from apscheduler.schedulers.background import BackgroundScheduler
def job_func(text):
    print(text)
scheduler = BackgroundScheduler()
# 在 2017-12-13 时刻运行一次 job_func 方法
scheduler .add_job(job_func, 'date', run_date=date(2017, 12, 13), args=['text'])
# 在 2017-12-13 14:00:00 时刻运行一次 job_func 方法
scheduler .add_job(job_func, 'date', run_date=datetime(2017, 12, 13, 14, 0, 0), args=['text'])
# 在 2017-12-13 14:00:01 时刻运行一次 job_func 方法
scheduler .add_job(job_func, 'date', run_date='2017-12-13 14:00:01', args=['text'])
scheduler.start()

2）interval 触发器
固定时间间隔触发。interval 间隔调度，参数如下：

interval 触发器使用示例如下：

import datetime
from apscheduler.schedulers.background import BackgroundScheduler
def job_func(text):
    print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])
scheduler = BackgroundScheduler()
# 每隔两分钟执行一次 job_func 方法
scheduler .add_job(job_func, 'interval', minutes=2)
# 在 2017-12-13 14:00:01 ~ 2017-12-13 14:00:10 之间, 每隔两分钟执行一次 job_func 方法
scheduler .add_job(job_func, 'interval', minutes=2, start_date='2017-12-13 14:00:01' , end_date='2017-12-13 14:00:10')
scheduler.start()

3）cron 触发器
在特定时间周期性地触发，和Linux crontab格式兼容。它是功能最强大的触发器。
我们先了解 cron 参数：

参照 https://www.cnblogs.com/ohyb/p/9084011.html
https://none.rclab.tk/guide/scheduler.html