Python定時任務框架APScheduler安裝使用詳解

前言

在日常工作中，如果想要簡化工作流程實現辦公自動化，那麼幾乎有大半的功能模組都需要使用定時任務，例如定時收發郵件，或者定時發微信或是檢測垃圾郵件等等，而在python中常用實現定時任務的包含以下四種方法：

• while True: + sleep()
• threading.Timer定時器
• 排程模組schedule
• 任務框架APScheduler

但在實際測試中，可以發現：

迴圈+sleep方式可以用來做簡單測試。
timer可以實現非同步定時任務。
schedule可以定點定時執行，但是仍然需要while Ture配合，而且佔用記憶體大。
APScheduler框架更加強大，可以直接在裡面新增定點與定時任務，無可挑剔！！！

所以接下來講介紹如何使用APscheduler實現python定時任務！！！

一、APscheduler簡介

APscheduler全稱Advanced Python Scheduler，作用為在指定的時間規則執行指定的作業，其是基於Quartz的一個Python定時任務框架，實現了Quartz的所有功能，使用起來十分方便。提供了基於日期、固定時間間隔以及crontab型別的任務，並且可以持久化任務。基於這些功能，我們可以很方便的實現一個python定時任務系統。

二、APscheduler安裝

我們通過Anaconda管理虛擬環境，並進行APscheduler庫的安裝。

pip install apscheduler

三、APscheduler組成部分

觸發器(trigger)：包含排程邏輯，每一個作業有它自己的觸發器，用於決定接下來哪一個作業會執行。除了他們自己初始設定意外，觸發器完全是無狀態的。

作業儲存(job store)：儲存被排程的作業，預設的作業儲存是簡單地把作業儲存在記憶體中，其他的作業儲存是將作業儲存在資料庫中。一個作業的資料講在儲存在持久化作業儲存時被序列化，並在載入時被反序列化。排程器不能分享同一個作業儲存。

執行器(executor)：處理作業的執行，他們通常通過在作業中提交制定的可呼叫物件到一個執行緒或者進城池來進行。當作業完成時，執行器將會通知排程器。

排程器(scheduler)：其他的組成部分。通常在應用只有一個排程器，應用的開發者通常不會直接處理作業儲存、排程器和觸發器，相反，排程器提供了處理這些的合適的介面。設定作業儲存和執行器可以在排程器中完成，例如新增、修改和移除作業。　

1.Job 作業

作用：

Job作為APScheduler最小執行單位。
建立Job時指定執行的函數，函數中所需引數，Job執行時的一些設定資訊。

構建說明:

id：指定作業的唯一ID
name：指定作業的名字
trigger：apscheduler定義的觸發器，用於確定Job的執行時間，根據設定的trigger規則，計算得到下次執行此job的
時間， 滿足時將會執行
executor：apscheduler定義的執行器，job建立時設定執行器的名字，根據字串你名字到scheduler獲取到執行此
job的 執行器，執行job指定的函數
max_instances：執行此job的最大範例數，executor執行job時，根據job的id來計算執行次數，根據設定的最大範例數
來確定是否可執行
next_run_time：Job下次的執行時間，建立Job時可以指定一個時間[datetime],不指定的話則預設根據trigger獲取觸
發時間
misfire_grace_time：Job的延遲執行時間，例如Job的計劃執行時間是21:00:00，但因服務重啟或其他原因導致
21:00:31才執行，如果設定此key為40,則該job會繼續執行，否則將會丟棄此job
coalesce：Job是否合併執行，是一個bool值。例如scheduler停止20s後重啟啟動，而job的觸發器設定為5s執行
一次，因此此job錯過了4個執行時間，如果設定為是，則會合併到一次執行，否則會逐個執行
func：Job執行的函數
args：Job執行函數需要的位置引數
kwargs：Job執行函數需要的關鍵字引數

2.Trigger 觸發器

包含排程邏輯，每一個作業有它自己的觸發器，用於決定接下來哪一個作業會執行。除了它們自己初始設定以外，觸發器完全是無狀態的。

APScheduler 有三種內建的 trigger:

•  date: 特定的時間點觸發
•  interval: 固定時間間隔觸發 
• cron: 在特定時間週期性地觸發

觸發器就是根據你指定的觸發方式，比如是按照時間間隔，還是按照 crontab觸發，觸發條件是什麼等。每個任務都有自己的觸發器。

3.Jobstore 作業儲存

如果你的應用在每次啟動的時候都會重新建立作業，那麼使用預設的作業記憶體（MemoryJobStore）即可，但是如果你需要在排程器重啟或者應用程式奔潰的情況下任然保留作業，你應該根據你的應用環境來選擇具體的作業記憶體。例如：使用Mongo或者SQLAlchemy JobStore （用於支援大多數RDBMS）。

任務記憶體是可以儲存任務的地方，預設情況下任務儲存在記憶體，也可將任務儲存在各種資料庫中。任務儲存進去後，會進行序列化，然後也可以反序列化提取出來，繼續執行。

4.Executor 執行器

Executor在scheduler中初始化，另外也可通過scheduler的add_executor動態新增Executor。 
每個executor都會繫結一個alias，這個作為唯一標識繫結到Job，在實際執行時會根據Job繫結的executor。找到實際的執行器物件，然後根據執行器物件執行Job。

Executor的種類會根據不同的排程來選擇，如果選擇AsyncIO作為排程的庫，那麼選擇AsyncIOExecutor，如果選擇tornado作為排程的庫，選擇TornadoExecutor，如果選擇啟動程序作為排程，選擇ThreadPoolExecutor或者ProcessPoolExecutor都可以。

Executor的選擇需要根據實際的scheduler來選擇不同的執行器。

處理作業的執行，它們通常通過在作業中提交制定的可呼叫物件到一個執行緒或者進城池來進行。當作業完成時，執行器將會通知排程器。

5.scheduler 排程器

Scheduler是APScheduler的核心，所有相關元件通過其定義。scheduler啟動之後，將開始按照設定的任務進行排程。除了依據所有定義Job的trigger生成的將要排程時間喚醒排程之外。當發生Job資訊變更時也會觸發排程。

scheduler可根據自身的需求選擇不同的元件，如果是使用AsyncIO則選擇AsyncIOScheduler，使用tornado則選擇TornadoScheduler。

任務排程器是屬於整個排程的總指揮官。它會合理安排作業記憶體、執行器、觸發器進行工作，並進行新增和刪除任務等。排程器通常是隻有一個的。開發人員很少直接操作觸發器、記憶體、執行器等。因為這些都由排程器自動來實現了。

四、Scheduler工作流程圖

1.Scheduler新增job流程

2.Scheduler排程流程 

五、APscheduler使用

1.簡單應用

import time
from apscheduler.schedulers.blocking import BlockingScheduler
def my_job(i):
    print (i)
sched = BlockingScheduler()
sched.add_job(my_job, 'interval', seconds=5,values=['學會了'])
sched.start()

 輸出：

2.操作作業

2.1 date觸發器

date 是最基本的一種排程，作業任務只會執行一次。它表示特定的時間點觸發。它的引數如下：

import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
def my_job(text):
    print(text)
# datetime型別（用於精確時間）
scheduler.add_job(my_job, 'date', run_date=datetime(2022, 4, 25, 17, 30, 5), args=['測試任務'])
 
scheduler.start()

注意：run_date引數可以是date型別、datetime型別或文字型別。

2.2 interval觸發器

固定時間間隔觸發。interval 間隔排程，引數如下：

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
def job_func():
     print("當前時間：", datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")
scheduler = BlockingScheduler()
# 每2小時觸發
scheduler.add_job(job_func, 'interval', hours=2)
# 在 2019-04-15 17:00:00 ~ 2019-12-31 24:00:00 之間, 每隔兩分鐘執行一次 job_func 方法
scheduler .add_job(job_func, 'interval', minutes=2, start_date='2022-04-29 17:00:00' , end_date='2022-12-31 24:00:00')
scheduler.start()

jitter振動引數，給每次觸發新增一個隨機浮動秒數，一般適用於多伺服器，避免同時執行造成服務擁堵。

# 每小時（上下浮動120秒區間內）執行`job_function`
scheduler.add_job(job_func, 'interval', hours=1, jitter=120)

 2.3 cron觸發器

在特定時間週期性地觸發，和Linux crontab格式相容。它是功能最強大的觸發器。

cron 引數：

表示式型別

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()
# 在每年 1-3、7-9 月份中的每個星期一、二中的 00:00, 01:00, 02:00 和 03:00 執行 job_func 任務
scheduler .add_job(job_func, 'cron', month='1-3,7-9',day='0, tue', hour='0-3')
scheduler.start()

參考：

python定時任務最強框架APScheduler詳細教學

python 定時任務APScheduler 使用介紹 

Python定時庫APScheduler原理及用法

到此這篇關於Python定時任務框架APScheduler安裝使用詳解的文章就介紹到這了,更多相關Python定時任務APScheduler內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！