前言

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

• while True: + sleep()
• threading.Timer定時器
• 調(diào)度模塊schedule
• 任務(wù)框架APScheduler

但在實際測試中，可以發(fā)現(xiàn)：

循環(huán)+sleep方式可以用來做簡單測試。
timer可以實現(xiàn)異步定時任務(wù)。
schedule可以定點(diǎn)定時執(zhí)行，但是仍然需要while Ture配合，而且占用內(nèi)存大。
APScheduler框架更加強(qiáng)大，可以直接在里面添加定點(diǎn)與定時任務(wù)，無可挑剔！！！

所以接下來講介紹如何使用APscheduler實現(xiàn)python定時任務(wù)！！！

一、APscheduler簡介

APscheduler全稱Advanced Python Scheduler，作用為在指定的時間規(guī)則執(zhí)行指定的作業(yè)，其是基于Quartz的一個Python定時任務(wù)框架，實現(xiàn)了Quartz的所有功能，使用起來十分方便。提供了基于日期、固定時間間隔以及crontab類型的任務(wù)，并且可以持久化任務(wù)。基于這些功能，我們可以很方便的實現(xiàn)一個python定時任務(wù)系統(tǒng)。

二、APscheduler安裝

我們通過Anaconda管理虛擬環(huán)境，并進(jìn)行APscheduler庫的安裝。

pip install apscheduler

三、APscheduler組成部分

觸發(fā)器(trigger)：包含調(diào)度邏輯，每一個作業(yè)有它自己的觸發(fā)器，用于決定接下來哪一個作業(yè)會運(yùn)行。除了他們自己初始配置意外，觸發(fā)器完全是無狀態(tài)的。

作業(yè)存儲(job store)：存儲被調(diào)度的作業(yè)，默認(rèn)的作業(yè)存儲是簡單地把作業(yè)保存在內(nèi)存中，其他的作業(yè)存儲是將作業(yè)保存在數(shù)據(jù)庫中。一個作業(yè)的數(shù)據(jù)講在保存在持久化作業(yè)存儲時被序列化，并在加載時被反序列化。調(diào)度器不能分享同一個作業(yè)存儲。

執(zhí)行器(executor)：處理作業(yè)的運(yùn)行，他們通常通過在作業(yè)中提交制定的可調(diào)用對象到一個線程或者進(jìn)城池來進(jìn)行。當(dāng)作業(yè)完成時，執(zhí)行器將會通知調(diào)度器。

調(diào)度器(scheduler)：其他的組成部分。通常在應(yīng)用只有一個調(diào)度器，應(yīng)用的開發(fā)者通常不會直接處理作業(yè)存儲、調(diào)度器和觸發(fā)器，相反，調(diào)度器提供了處理這些的合適的接口。配置作業(yè)存儲和執(zhí)行器可以在調(diào)度器中完成，例如添加、修改和移除作業(yè)。　

1.Job 作業(yè)

作用：

Job作為APScheduler最小執(zhí)行單位。
創(chuàng)建Job時指定執(zhí)行的函數(shù)，函數(shù)中所需參數(shù)，Job執(zhí)行時的一些設(shè)置信息。

構(gòu)建說明:

id：指定作業(yè)的唯一ID
name：指定作業(yè)的名字
trigger：apscheduler定義的觸發(fā)器，用于確定Job的執(zhí)行時間，根據(jù)設(shè)置的trigger規(guī)則，計算得到下次執(zhí)行此job的
時間， 滿足時將會執(zhí)行
executor：apscheduler定義的執(zhí)行器，job創(chuàng)建時設(shè)置執(zhí)行器的名字，根據(jù)字符串你名字到scheduler獲取到執(zhí)行此
job的 執(zhí)行器，執(zhí)行job指定的函數(shù)
max_instances：執(zhí)行此job的最大實例數(shù)，executor執(zhí)行job時，根據(jù)job的id來計算執(zhí)行次數(shù)，根據(jù)設(shè)置的最大實例數(shù)
來確定是否可執(zhí)行
next_run_time：Job下次的執(zhí)行時間，創(chuàng)建Job時可以指定一個時間[datetime],不指定的話則默認(rèn)根據(jù)trigger獲取觸
發(fā)時間
misfire_grace_time：Job的延遲執(zhí)行時間，例如Job的計劃執(zhí)行時間是21:00:00，但因服務(wù)重啟或其他原因?qū)е?br /> 21:00:31才執(zhí)行，如果設(shè)置此key為40,則該job會繼續(xù)執(zhí)行，否則將會丟棄此job
coalesce：Job是否合并執(zhí)行，是一個bool值。例如scheduler停止20s后重啟啟動，而job的觸發(fā)器設(shè)置為5s執(zhí)行
一次，因此此job錯過了4個執(zhí)行時間，如果設(shè)置為是，則會合并到一次執(zhí)行，否則會逐個執(zhí)行
func：Job執(zhí)行的函數(shù)
args：Job執(zhí)行函數(shù)需要的位置參數(shù)
kwargs：Job執(zhí)行函數(shù)需要的關(guān)鍵字參數(shù)

2.Trigger 觸發(fā)器

包含調(diào)度邏輯，每一個作業(yè)有它自己的觸發(fā)器，用于決定接下來哪一個作業(yè)會運(yùn)行。除了它們自己初始配置以外，觸發(fā)器完全是無狀態(tài)的。

APScheduler 有三種內(nèi)建的 trigger:

• ?date: 特定的時間點(diǎn)觸發(fā)
• ?interval: 固定時間間隔觸發(fā)
• ?cron: 在特定時間周期性地觸發(fā)

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

3.Jobstore 作業(yè)存儲

如果你的應(yīng)用在每次啟動的時候都會重新創(chuàng)建作業(yè)，那么使用默認(rèn)的作業(yè)存儲器（MemoryJobStore）即可，但是如果你需要在調(diào)度器重啟或者應(yīng)用程序奔潰的情況下任然保留作業(yè)，你應(yīng)該根據(jù)你的應(yīng)用環(huán)境來選擇具體的作業(yè)存儲器。例如：使用Mongo或者SQLAlchemy JobStore （用于支持大多數(shù)RDBMS）。

任務(wù)存儲器是可以存儲任務(wù)的地方，默認(rèn)情況下任務(wù)保存在內(nèi)存，也可將任務(wù)保存在各種數(shù)據(jù)庫中。任務(wù)存儲進(jìn)去后，會進(jìn)行序列化，然后也可以反序列化提取出來，繼續(xù)執(zhí)行。

4.Executor 執(zhí)行器

Executor在scheduler中初始化，另外也可通過scheduler的add_executor動態(tài)添加Executor。?
每個executor都會綁定一個alias，這個作為唯一標(biāo)識綁定到Job，在實際執(zhí)行時會根據(jù)Job綁定的executor。找到實際的執(zhí)行器對象，然后根據(jù)執(zhí)行器對象執(zhí)行Job。

Executor的種類會根據(jù)不同的調(diào)度來選擇，如果選擇AsyncIO作為調(diào)度的庫，那么選擇AsyncIOExecutor，如果選擇tornado作為調(diào)度的庫，選擇TornadoExecutor，如果選擇啟動進(jìn)程作為調(diào)度，選擇ThreadPoolExecutor或者ProcessPoolExecutor都可以。

Executor的選擇需要根據(jù)實際的scheduler來選擇不同的執(zhí)行器。

處理作業(yè)的運(yùn)行，它們通常通過在作業(yè)中提交制定的可調(diào)用對象到一個線程或者進(jìn)城池來進(jìn)行。當(dāng)作業(yè)完成時，執(zhí)行器將會通知調(diào)度器。

5.scheduler 調(diào)度器

Scheduler是APScheduler的核心，所有相關(guān)組件通過其定義。scheduler啟動之后，將開始按照配置的任務(wù)進(jìn)行調(diào)度。除了依據(jù)所有定義Job的trigger生成的將要調(diào)度時間喚醒調(diào)度之外。當(dāng)發(fā)生Job信息變更時也會觸發(fā)調(diào)度。

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

任務(wù)調(diào)度器是屬于整個調(diào)度的總指揮官。它會合理安排作業(yè)存儲器、執(zhí)行器、觸發(fā)器進(jìn)行工作，并進(jìn)行添加和刪除任務(wù)等。調(diào)度器通常是只有一個的。開發(fā)人員很少直接操作觸發(fā)器、存儲器、執(zhí)行器等。因為這些都由調(diào)度器自動來實現(xiàn)了。

四、Scheduler工作流程圖

1.Scheduler添加job流程

2.Scheduler調(diào)度流程?

?

五、APscheduler使用

1.簡單應(yīng)用

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

?輸出：

2.操作作業(yè)

2.1?date觸發(fā)器

date?是最基本的一種調(diào)度，作業(yè)任務(wù)只會執(zhí)行一次。它表示特定的時間點(diǎn)觸發(fā)。它的參數(shù)如下：

參數(shù)	說明
run_date(datetime or str)	任務(wù)運(yùn)行的日期或者時間
timezone(datetime.tzinfo or str)	指定時區(qū)

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=['測試任務(wù)'])scheduler.start()

注意：run_date參數(shù)可以是date類型、datetime類型或文本類型。

2.2?interval觸發(fā)器

固定時間間隔觸發(fā)。interval 間隔調(diào)度，參數(shù)如下：

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

from datetime import datetime from apscheduler.schedulers.blocking import BlockingSchedulerdef job_func():print("當(dāng)前時間：", datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")scheduler = BlockingScheduler()# 每2小時觸發(fā) scheduler.add_job(job_func, 'interval', hours=2)# 在 2019-04-15 17:00:00 ~ 2019-12-31 24:00:00 之間, 每隔兩分鐘執(zhí)行一次 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振動參數(shù)，給每次觸發(fā)添加一個隨機(jī)浮動秒數(shù)，一般適用于多服務(wù)器，避免同時運(yùn)行造成服務(wù)擁堵。

# 每小時（上下浮動120秒?yún)^(qū)間內(nèi)）運(yùn)行`job_function` scheduler.add_job(job_func, 'interval', hours=1, jitter=120)

?2.3 cron觸發(fā)器

在特定時間周期性地觸發(fā)，和Linux crontab格式兼容。它是功能最強(qiáng)大的觸發(fā)器。

cron 參數(shù)：

參數(shù)	說明
year(int or str)	年，4位數(shù)字
month(int or str)	月（范圍1-12）
day(int or str)	日（范圍1-31）
week(int or str)	周（范圍1-53）
day_of_week(int or str)	周內(nèi)第幾天或者星期幾（范圍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)	最晚結(jié)束日期（含）
timezone(datetime.tzinfo or ? str)	指定時區(qū)

表達(dá)式類型

表達(dá)式參數(shù)類型描述

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

import datetime from apscheduler.schedulers.background import BackgroundSchedulerdef job_func(text):print("當(dāng)前時間：", 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 執(zhí)行 job_func 任務(wù) scheduler .add_job(job_func, 'cron', month='1-3,7-9',day='0, tue', hour='0-3')scheduler.start()

參考：

python定時任務(wù)最強(qiáng)框架APScheduler詳細(xì)教程_jspython的博客-CSDN博客_apscheduler python

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

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。