前言

在日常工作中，如果想要簡化工作流程實現辦公自動化，那么幾乎有大半的功能模塊都需要使用定時任務，例如定時收發郵件，或者定時發微信或是檢測垃圾郵件等等，而在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)：其他的組成部分。通常在應用只有一個調度器，應用的開發者通常不會直接處理作業存儲、調度器和觸發器，相反，調度器提供了處理這些的合適的接口。配置作業存儲和執行器可以在調度器中完成，例如添加、修改和移除作業?！?/p>

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 BlockingSchedulerdef my_job(i):print (i)sched = BlockingScheduler() sched.add_job(my_job, 'interval', seconds=5,values=['學會了']) sched.start()

?輸出：

2.操作作業

2.1?date觸發器

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

參數	說明
run_date(datetime or str)	任務運行的日期或者時間
timezone(datetime.tzinfo or str)	指定時區

import datetime from apscheduler.schedulers.blocking import BlockingSchedulerscheduler = 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 間隔調度，參數如下：

參數	說明
weeks(int)	間隔幾周
days(int)	間隔幾天
hours(int)	間隔幾小時
minutes(int)	間隔幾分鐘
seconds(int)	間隔多少秒
start_date(datetime or str)	開始日期
end_date(datetime or str)	結束日期
timezone(datetime.tzinfo or ? str)	時區

from datetime import datetime from apscheduler.schedulers.blocking import BlockingSchedulerdef 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 參數：

參數	說明
year(int or str)	年，4位數字
month(int or str)	月（范圍1-12）
day(int or str)	日（范圍1-31）
week(int or str)	周（范圍1-53）
day_of_week(int or str)	周內第幾天或者星期幾（范圍0-6或者mon,tue,wed,thu,fri,stat,sun）
hour(int or str)	時（0-23）
minute(int or str)	分（0-59）
second(int or str)	秒（0-59）
start_date(datetime or str)	最早開始日期（含）
end_date(datetime or str)	最晚結束日期（含）
timezone(datetime.tzinfo or ? str)	指定時區

表達式類型

表達式參數類型描述

*	所有	通配符。例：minutes=*即每分鐘觸發
*/a	所有	可被a整除的通配符。
a-b	所有	范圍a-b觸發
a-b/c	所有	范圍a-b，且可被c整除時觸發
xth y	日	第幾個星期幾觸發。x為第幾個，y為星期幾
last x	日	一個月中，最后的星期幾觸發
last	日	一個月最后一天觸發
x,y,z	所有	組合表達式，可以組合確定值或上方的表達式

import datetime from apscheduler.schedulers.background import BackgroundSchedulerdef 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詳細教程_jspython的博客-CSDN博客_apscheduler python

python 定時任務APScheduler 使用介紹 - 蒼青浪 - 博客園

Python定時庫APScheduler原理及用法 - 知乎

總結

以上是生活随笔為你收集整理的Python定时任务框架APScheduler详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。