1、簡(jiǎn)介

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

2、APScheduler四個(gè)組件

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

觸發(fā)器(trigger)

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

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

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

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

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

執(zhí)行器(executor)

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

最常用的 executor 有兩種：

ProcessPoolExecutor ThreadPoolExecutor

調(diào)度器(scheduler)

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

2、安裝

$ pip install apscheduler

接下來(lái)我們看下簡(jiǎn)單的幾個(gè)示例：

===============interval: 固定時(shí)間間隔觸發(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: 特定時(shí)間周期性地觸發(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點(diǎn)，每隔 1分鐘 運(yùn)行一次 job 方法scheduler.add_job(job, ’cron’, hour=17, minute=’*/1’, args=[’job1’])# 在每天22和23點(diǎn)的25分，運(yùn)行一次 job 方法scheduler.add_job(job, ’cron’, hour=’22-23’, minute=’25’, args=[’job2’])scheduler.start()

通過(guò)裝飾器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()

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