Python使用APScheduler實現定時任務

APScheduler是基於Quartz的一個Python定時任務框架。提供了基於日期、固定時間間隔以及crontab類型的任務，並且可以持久化任務。
線上文檔：https://apscheduler.readthedocs.io/en/latest/userguide.html

一、安裝APScheduler

pip install apscheduler

二、基本概念

APScheduler有四大組件：
1、觸發器 triggers ：
觸發器包含調度邏輯。每個作業都有自己的觸發器，用於確定下一個任務何時運行。除了初始配置之外，觸發器是完全無狀態的。
有三種內建的trigger:
（1）date: 特定的時間點觸發
（2）interval: 固定時間間隔觸發
（3）cron: 在特定時間周期性地觸發
2、任務儲存器 job stores：用於存放任務，把任務存放在記憶體（為預設MemoryJobStore）或資料庫中。
3、執行器 executors： 執行器是將任務提交到線程池或進程池中運行，當任務完成時，執行器通知調度器觸發相應的事件。
4、調度器 schedulers： 把上方三個組件作為參數，通過創建調度器實例來運行
根據開發需求選擇相應的組件，下麵是不同的調度器組件：
BlockingScheduler 阻塞式調度器：適用於只跑調度器的程式。
BackgroundScheduler 後臺調度器：適用於非阻塞的情況，調度器會在後臺獨立運行。
AsyncIOScheduler AsyncIO調度器，適用於應用使用AsnycIO的情況。
GeventScheduler Gevent調度器，適用於應用通過Gevent的情況。
TornadoScheduler Tornado調度器，適用於構建Tornado應用。
TwistedScheduler Twisted調度器，適用於構建Twisted應用。
QtScheduler Qt調度器，適用於構建Qt應用。

三、使用步驟
1、新建一個調度器schedulers
2、添加調度任務
3、運行調度任務

四、使用實例

1、觸發器date

特定的時間點觸發，只執行一次。參數如下：

使用例子：

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

def job(text):    
    print(text)

scheduler = BlockingScheduler()
# 在 2019-8-30 運行一次 job 方法
scheduler.add_job(job, 'date', run_date=date(2019, 8, 30), args=['text1'])
# 在 2019-8-30 01:00:00 運行一次 job 方法
scheduler.add_job(job, 'date', run_date=datetime(2019, 8, 30, 1, 0, 0), args=['text2'])
# 在 2019-8-30 01:00:01 運行一次 job 方法
scheduler.add_job(job, 'date', run_date='2019-8-30 01:00:00', args=['text3'])

scheduler.start()

2、觸發器interval

固定時間間隔觸發。參數如下：

使用例子：

import time
from apscheduler.schedulers.blocking import BlockingScheduler

def job(text):    
    t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    print('{} --- {}'.format(text, t))

scheduler = BlockingScheduler()
# 每隔 1分鐘 運行一次 job 方法
scheduler.add_job(job, 'interval', minutes=1, args=['job1'])
# 在 2019-08-29 22:15:00至2019-08-29 22:17:00期間，每隔1分30秒 運行一次 job 方法
scheduler.add_job(job, 'interval', minutes=1, seconds = 30, start_date='2019-08-29 22:15:00', end_date='2019-08-29 22:17:00', args=['job2'])

scheduler.start()

'''
運行結果：
job2 --- 2019-08-29 22:15:00
job1 --- 2019-08-29 22:15:46
job2 --- 2019-08-29 22:16:30
job1 --- 2019-08-29 22:16:46
job1 --- 2019-08-29 22:17:46
...餘下省略...
'''

 3、觸發器cron

在特定時間周期性地觸發。參數如下：

這些參數支持算數表達式，取值格式有如下：

  使用例子：

import time
from apscheduler.schedulers.blocking import BlockingScheduler

def job(text):    
    t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    print('{} --- {}'.format(text, t))

scheduler = BlockingScheduler()
# 在每天22點，每隔 1分鐘 運行一次 job 方法
scheduler.add_job(job, 'cron', hour=22, minute='*/1', args=['job1'])
# 在每天22和23點的25分，運行一次 job 方法
scheduler.add_job(job, 'cron', hour='22-23', minute='25', args=['job2'])

scheduler.start()

'''
運行結果：
job1 --- 2019-08-29 22:25:00
job2 --- 2019-08-29 22:25:00
job1 --- 2019-08-29 22:26:00
job1 --- 2019-08-29 22:27:00
...餘下省略...
'''

 4、通過裝飾器scheduled_job()添加方法

 添加任務的方法有兩種：

（1）通過調用add_job()---見上面1至3代碼
（2）通過裝飾器scheduled_job()：
第一種方法是最常用的方法。第二種方法主要是方便地聲明在應用程式運行時不會更改的任務。該 add_job()方法返回一個apscheduler.job.Job實例，可以使用該實例稍後修改或刪除該任務。

import time
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job('interval', seconds=5)
def job1():    
    t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    print('job1 --- {}'.format(t))

@scheduler.scheduled_job('cron', second='*/7')
def job2():    
    t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    print('job2 --- {}'.format(t))

scheduler.start()

'''
運行結果：
job2 --- 2019-08-29 22:36:35
job1 --- 2019-08-29 22:36:37
job2 --- 2019-08-29 22:36:42
job1 --- 2019-08-29 22:36:42
job1 --- 2019-08-29 22:36:47
job2 --- 2019-08-29 22:36:49
...餘下省略...
'''