一、APScheduler簡介

APScheduler（Advanced Python Scheduler）是一個輕量級的Python定時任務(wù)調(diào)度框架（Python庫）。APScheduler有三個內(nèi)置的調(diào)度系統(tǒng)，其中包括：

• cron式調(diào)度（可選開始/結(jié)束時間）

• 基于間隔的執(zhí)行（以偶數(shù)間隔運(yùn)行作業(yè)，也可以選擇開始/結(jié)束時間）

• 一次性延遲執(zhí)行任務(wù)（在指定的日期/時間內(nèi)運(yùn)行作業(yè)一次）

支持的后端存儲作業(yè)

APScheduler可以任意混合和匹配調(diào)度系統(tǒng)和作業(yè)存儲的后端，其中支持后端存儲作業(yè)包括：

• Memory

• SQLAlchemy

• MongoDB

• Redis

• RethinkDB

• ZooKeeper

集成的Python框架

APScheduler內(nèi)繼承了幾個常見的Python框架：

• asyncio

• gevent

• tornado

• qt

二、APScheduler下載安裝

使用pip安裝：

pip?install?apscheduler pip?install?apscheduler==3.6.3

如果超時或者出現(xiàn)別的情況，可以選擇：

#?法1使用豆瓣源下載 pip?install?-i?https://pypi.doubanio.com/simple/?apscheduler #?法2使用清華源下載 pip?install?-i?https://pypi.tuna.tsinghua.edu.cn/simple?apscheduler

要是再不行，點(diǎn)擊該鏈接或者pypi官網(wǎng)下載了。下載并解壓縮，進(jìn)入跟setup.py文件同級的目錄，打開cmd，使用命令進(jìn)行下載：

python?setup.py?install

三、APScheduler組件

APScheduler共有4種組件，分別是：

• 觸發(fā)器（trigger），觸發(fā)器中包含調(diào)度邏輯，每個作業(yè)都有自己的觸發(fā)器來決定下次運(yùn)行時間。除了它們自己初始配置以外，觸發(fā)器完全是無狀態(tài)的。

• 作業(yè)存儲器（job store），存儲被調(diào)度的作業(yè)，默認(rèn)的作業(yè)存儲器只是簡單地把作業(yè)保存在內(nèi)存中，其他的作業(yè)存儲器則是將作業(yè)保存在數(shù)據(jù)庫中，當(dāng)作業(yè)被保存在一個持久化的作業(yè)存儲器中的時候，該作業(yè)的數(shù)據(jù)會被序列化，并在加載時被反序列化，需要說明的是，作業(yè)存儲器不能共享調(diào)度器。

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

• 調(diào)度器（scheduler），配置作業(yè)存儲器和執(zhí)行器可以在調(diào)度器中完成。例如添加、修改、移除作業(yè)，根據(jù)不同的應(yīng)用場景，可以選擇不同的調(diào)度器，可選的將在下一小節(jié)展示。

各組件簡介

調(diào)度器

• BlockingScheduler : 當(dāng)調(diào)度器是你應(yīng)用中唯一要運(yùn)行的東西時。

• BackgroundScheduler : 當(dāng)你沒有運(yùn)行任何其他框架并希望調(diào)度器在你應(yīng)用的后臺執(zhí)行時使用（充電樁即使用此種方式）。

• AsyncIOScheduler : 當(dāng)你的程序使用了asyncio（一個異步框架）的時候使用。

• GeventScheduler : 當(dāng)你的程序使用了gevent（高性能的Python并發(fā)框架）的時候使用。

• TornadoScheduler : 當(dāng)你的程序基于Tornado（一個web框架）的時候使用。

• TwistedScheduler : 當(dāng)你的程序使用了Twisted（一個異步框架）的時候使用

• QtScheduler : 如果你的應(yīng)用是一個Qt應(yīng)用的時候可以使用。

作業(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）

執(zhí)行器

對執(zhí)行器的選擇取決于你使用上面哪些框架，大多數(shù)情況下，使用默認(rèn)的ThreadPoolExecutor已經(jīng)能夠滿足需求。如果你的應(yīng)用涉及到CPU密集型操作，你可以考慮使用ProcessPoolExecutor來使用更多的CPU核心。你也可以同時使用兩者，將ProcessPoolExecutor作為第二執(zhí)行器。

觸發(fā)器

當(dāng)你調(diào)度作業(yè)的時候，你需要為這個作業(yè)選擇一個觸發(fā)器，用來描述這個作業(yè)何時被觸發(fā)，APScheduler有三種內(nèi)置的觸發(fā)器類型：

• date 一次性指定日期

• interval 在某個時間范圍內(nèi)間隔多長時間執(zhí)行一次

• cron 和Linux crontab格式兼容，最為強(qiáng)大

四、使用

當(dāng)你需要調(diào)度作業(yè)的時候，你需要為這個作業(yè)選擇一個觸發(fā)器，用來描述該作業(yè)將在何時被觸發(fā)，APScheduler有3中內(nèi)置的觸發(fā)器類型：

• 新建一個調(diào)度器（scheduler）

• 添加一個調(diào)度任務(wù)（job store)

• 運(yùn)行調(diào)度任務(wù)

添加任務(wù)

有兩種方式可以添加一個新的作業(yè)：

• add_job來添加作業(yè)

• 裝飾器模式添加作業(yè)

指定時間執(zhí)行任務(wù)，只執(zhí)行一次

import?datetime from?apscheduler.schedulers.blocking?import?BlockingScheduler def?job2(text):print('job2',?datetime.datetime.now(),?text) scheduler?=?BlockingScheduler() scheduler.add_job(job2,?'date',?run_date=datetime.datetime(2020,?2,?25,?19,?5,?6),?args=['text'],?id='job2') scheduler.start()

上例中，只在2010-2-25 19:05:06執(zhí)行一次，args傳遞一個text參數(shù)。

間隔時間執(zhí)行任務(wù)

下面來個簡單的例子，作業(yè)每個5秒執(zhí)行一次：

import?datetime from?apscheduler.schedulers.blocking?import?BlockingSchedulerdef?job1():print('job1',?datetime.datetime.now()) scheduler?=?BlockingScheduler() scheduler.add_job(job1,?'interval',?seconds=5,?id='job1')??#?每隔5秒執(zhí)行一次 scheduler.start()

每天凌晨1點(diǎn)30分50秒執(zhí)行一次

#?裝飾器的方式from?apscheduler.schedulers.blocking?import?BlockingScheduler??#?后臺運(yùn)行 sc?=?BlockingScheduler() f?=?open('t1.text',?'a',?encoding='utf8')@sc.scheduled_job('cron',?day_of_week='*',?hour=1,?minute='30',?second='50') def?check_db():print(111111111111) if?__name__?==?'__main__':try:sc.start()f.write('定時任務(wù)成功執(zhí)行')except?Exception?as?e:sc.shutdown()f.write('定時任務(wù)執(zhí)行失敗')finally:f.close()

每幾分鐘執(zhí)行一次：

import?datetime from?apscheduler.schedulers.blocking?import?BlockingSchedulerdef?job1():print('job1',?datetime.datetime.now()) scheduler?=?BlockingScheduler() #?每隔2分鐘執(zhí)行一次，?*/1：每隔1分鐘執(zhí)行一次 scheduler.add_job(job1,?'cron',?minute="*/2",?id='job1')? scheduler.start()

每小時執(zhí)行一次：

import?datetime from?apscheduler.schedulers.blocking?import?BlockingSchedulerdef?job1():print('job1',?datetime.datetime.now()) scheduler?=?BlockingScheduler() #?每小時執(zhí)行一次 scheduler.add_job(job1,?'interval',?hours=1,?id='job1') #?每小時執(zhí)行一次，上下浮動120秒?yún)^(qū)間內(nèi) #?scheduler.add_job(job1,?'interval',?hours=1,?id='job1',?jitter=120) scheduler.start()

總結(jié)

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

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