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
特定的時間點觸發，只執行一次。引數如下：
| 引數 | 說明 |
| run_date (datetime 或 str) | 作業的執行日期或時間 |
| timezone (datetime.tzinfo 或 str) | 指定時區 |
使用例子：
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,run_date=datetime(2019,30,1,0),args=['text2']) # 在 2019-8-30 01:00:01 執行一次 job 方法 scheduler.add_job(job,run_date='2019-8-30 01:00:00',args=['text3']) scheduler.start()
2、觸發器interval
固定時間間隔觸發。引數如下：
| 引數 | 說明 |
| weeks (int) | 間隔幾周 |
| days (int) | 間隔幾天 |
| hours (int) | 間隔幾小時 |
| minutes (int) | 間隔幾分鐘 |
| seconds (int) | 間隔多少秒 |
| start_date (datetime 或 str) | 開始日期 |
| end_date (datetime 或 str) | 結束日期 |
| timezone (datetime.tzinfo 或str) |
使用例子：
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,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
在特定時間週期性地觸發。引數如下：
| 引數 | 說明 |
| year (int 或 str) | 年，4位數字 |
| month (int 或 str) | 月 (範圍1-12) |
| day (int 或 str) | 日 (範圍1-31) |
| week (int 或 str) | 周 (範圍1-53) |
| day_of_week (int 或 str) | 周內第幾天或者星期幾 (範圍0-6 或者 mon,tue,wed,thu,fri,sat,sun) |
| hour (int 或 str) | 時 (範圍0-23) |
| minute (int 或 str) | 分 (範圍0-59) |
| second (int 或 str) | 秒 (範圍0-59) |
| start_date (datetime 或 str) | 最早開始日期(包含) |
| end_date (datetime 或 str) | 最晚結束時間(包含) |
| timezone (datetime.tzinfo 或str) | 指定時區 |
這些引數支援算數表示式，取值格式有如下：
使用例子：
import time
from apscheduler.schedulers.blocking import BlockingScheduler
def job(text):
t = time.strftime('%Y-%m-%d %H:%M:%S',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,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
...餘下省略...
'''
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。