1、簡介

APScheduler是一個 Python 定時任務(wù)框架，使用起來十分方便。提供了基于日期、固定時間間隔以及 crontab 類型的任務(wù)，并且可以持久化任務(wù)、并以 daemon 方式運行應(yīng)用。

2、APScheduler四個組件

APScheduler 四個組件分別為：觸發(fā)器(trigger)，作業(yè)存儲(job store)，執(zhí)行器(executor)，調(diào)度器(scheduler)。

觸發(fā)器(trigger)

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

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

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

作業(yè)存儲(job store)

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

執(zhí)行器(executor)

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

最常用的 executor 有兩種：

ProcessPoolExecutor ThreadPoolExecutor

調(diào)度器(scheduler)

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

2、安裝

$ pip install apscheduler

接下來我們看下簡單的幾個示例：

===============interval: 固定時間間隔觸發(fā)===============from apscheduler.schedulers.blocking import BlockingSchedulerfrom datetime import datetimedef job(): print(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))# 定義BlockingSchedulersched = BlockingScheduler()sched.add_job(job, ’interval’, seconds=5) sched.start()

===============cron: 特定時間周期性地觸發(fā)===============import timefrom apscheduler.schedulers.blocking import BlockingSchedulerdef 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=17, minute=’*/1’, args=[’job1’])# 在每天22和23點的25分，運行一次 job 方法scheduler.add_job(job, ’cron’, hour=’22-23’, minute=’25’, args=[’job2’])scheduler.start()

通過裝飾器scheduled_job()添加方法

import timefrom apscheduler.schedulers.blocking import BlockingSchedulerscheduler = 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()

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。