這篇文章將為大家詳細(xì)講解有關(guān)怎么在Python中利用APScheduler實現(xiàn)一個定時任務(wù)，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關(guān)知識有一定的了解。
一、安裝APScheduler
pip install apscheduler
二、基本概念
APScheduler有四大組件：
1、觸發(fā)器 triggers ：
觸發(fā)器包含調(diào)度邏輯。每個作業(yè)都有自己的觸發(fā)器，用于確定下一個任務(wù)何時運行。除了初始配置之外，觸發(fā)器是完全無狀態(tài)的。
有三種內(nèi)建的trigger:
（1）date: 特定的時間點觸發(fā)
（2）interval: 固定時間間隔觸發(fā)
（3）cron: 在特定時間周期性地觸發(fā)
2、任務(wù)儲存器 job stores：用于存放任務(wù)，把任務(wù)存放在內(nèi)存（為默認(rèn)MemoryJobStore）或數(shù)據(jù)庫中。
3、執(zhí)行器 executors： 執(zhí)行器是將任務(wù)提交到線程池或進(jìn)程池中運行，當(dāng)任務(wù)完成時，執(zhí)行器通知調(diào)度器觸發(fā)相應(yīng)的事件。
4、調(diào)度器 schedulers： 把上方三個組件作為參數(shù)，通過創(chuàng)建調(diào)度器實例來運行
根據(jù)開發(fā)需求選擇相應(yīng)的組件，下面是不同的調(diào)度器組件：
BlockingScheduler 阻塞式調(diào)度器：適用于只跑調(diào)度器的程序。
BackgroundScheduler 后臺調(diào)度器：適用于非阻塞的情況，調(diào)度器會在后臺獨立運行。
AsyncIOScheduler AsyncIO調(diào)度器，適用于應(yīng)用使用AsnycIO的情況。
GeventScheduler Gevent調(diào)度器，適用于應(yīng)用通過Gevent的情況。
TornadoScheduler Tornado調(diào)度器，適用于構(gòu)建Tornado應(yīng)用。
TwistedScheduler Twisted調(diào)度器，適用于構(gòu)建Twisted應(yīng)用。
QtScheduler Qt調(diào)度器，適用于構(gòu)建Qt應(yīng)用。
三、使用步驟
1、新建一個調(diào)度器schedulers
2、添加調(diào)度任務(wù)
3、運行調(diào)度任務(wù)
四、使用實例
1、觸發(fā)器date
特定的時間點觸發(fā)，只執(zhí)行一次。參數(shù)如下：
| 參數(shù) | 說明 |
| run_date (datetime 或 str) | 作業(yè)的運行日期或時間 |
| timezone (datetime.tzinfo 或 str) | 指定時區(qū) |
使用例子：
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、觸發(fā)器interval
固定時間間隔觸發(fā)。參數(shù)如下：
| 參數(shù) | 說明 |
| weeks (int) | 間隔幾周 |
| days (int) | 間隔幾天 |
| hours (int) | 間隔幾小時 |
| minutes (int) | 間隔幾分鐘 |
| seconds (int) | 間隔多少秒 |
| start_date (datetime 或 str) | 開始日期 |
| end_date (datetime 或 str) | 結(jié)束日期 |
| 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, '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()
'''
運行結(jié)果：
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、觸發(fā)器cron
在特定時間周期性地觸發(fā)。參數(shù)如下：
| 參數(shù) | 說明 |
| year (int 或 str) | 年，4位數(shù)字 |
| month (int 或 str) | 月 (范圍1-12) |
| day (int 或 str) | 日 (范圍1-31) |
| week (int 或 str) | 周 (范圍1-53) |
| day_of_week (int 或 str) | 周內(nèi)第幾天或者星期幾 (范圍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) | 最晚結(jié)束時間(包含) |
| timezone (datetime.tzinfo 或str) | 指定時區(qū) |
這些參數(shù)支持算數(shù)表達(dá)式，取值格式有如下：
使用例子：
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()
'''
運行結(jié)果：
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()添加方法
添加任務(wù)的方法有兩種：
（1）通過調(diào)用add_job()---見上面1至3代碼
（2）通過裝飾器scheduled_job()：
第一種方法是最常用的方法。第二種方法主要是方便地聲明在應(yīng)用程序運行時不會更改的任務(wù)。該 add_job()方法返回一個apscheduler.job.Job實例，可以使用該實例稍后修改或刪除該任務(wù)。
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()
'''
運行結(jié)果：
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
...余下省略...
'''關(guān)于怎么在Python中利用APScheduler實現(xiàn)一個定時任務(wù)就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，可以學(xué)到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。
另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機、免備案服務(wù)器”等云主機租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。
