導(dǎo)語(yǔ)

在工作場(chǎng)景遇到了這么一個(gè)場(chǎng)景，就是需要定期去執(zhí)行一個(gè)緩存接口，用于同步設(shè)備配置。首先想到的就是Linux上的crontab，可以定期，或者間隔一段時(shí)間去執(zhí)行任務(wù)。但是如果你想要把這個(gè)定時(shí)任務(wù)作為一個(gè)模塊集成到Python項(xiàng)目中，或者想持久化任務(wù)，顯然crontab不太適用。Python的APScheduler模塊能夠很好的解決此類(lèi)問(wèn)題，所以專(zhuān)門(mén)寫(xiě)這篇文章，從簡(jiǎn)單入門(mén)開(kāi)始記錄關(guān)于A(yíng)PScheduler最基礎(chǔ)的使用場(chǎng)景，以及解決持久化任務(wù)的問(wèn)題，最后結(jié)合其他框架深層次定制定時(shí)任務(wù)模塊這幾個(gè)點(diǎn)入手。

簡(jiǎn)單介紹

先簡(jiǎn)單介紹一下Apscheduler模塊包含的四種組件：

Trigger觸發(fā)器 Job作業(yè) Excutor執(zhí)行器 Scheduler調(diào)度器

大概了解了Apscheduler包含的幾種概念，現(xiàn)在先來(lái)看一下一個(gè)簡(jiǎn)單的示例：

# -*- coding: utf-8 -*-from apscheduler.schedulers.blocking import BlockingSchedulerimport timedef hello(): print(time.strftime('%c'))if __name__ == '__main__': scheduler = BlockingScheduler() scheduler.add_job(hello, ’interval’, seconds=5) scheduler.start()

示例的輸出：

Thu Dec 3 16:01:20 2020Thu Dec 3 16:01:25 2020Thu Dec 3 16:01:30 2020Thu Dec 3 16:01:35 2020Thu Dec 3 16:01:40 2020..........

這個(gè)簡(jiǎn)單的示例，我們用上面提到幾種組件分析一下運(yùn)行邏輯：

首先是Scheduler調(diào)度器，這個(gè)示例使用的BlockingScheduler調(diào)度器，在官方文檔中的解釋是，BlockingScheduler適合當(dāng)你的這個(gè)定時(shí)任務(wù)程序是唯一運(yùn)行的程序；換言之，則是BlockingScheduler調(diào)度器是一個(gè)阻塞調(diào)度器，當(dāng)程序運(yùn)行這種調(diào)度器，進(jìn)程則會(huì)阻塞，無(wú)法執(zhí)行其他操作； 其次是Job作業(yè)和觸發(fā)器，這兩個(gè)放在一起講是因?yàn)椋诙x作業(yè)的時(shí)候，你就需要選擇一個(gè)觸發(fā)器，這里選擇的是interval觸發(fā)器，這種觸發(fā)器會(huì)以固定時(shí)間間隔運(yùn)行作業(yè)。換言之，為調(diào)度器添加一個(gè)hello的工作，并以每5秒的時(shí)間間隔執(zhí)行任務(wù)。 最后就是執(zhí)行器，默認(rèn)是ThreadPoolExcutor執(zhí)行器，他們將任務(wù)中可調(diào)用對(duì)象交給線(xiàn)程池執(zhí)行操作，等完成操作后，執(zhí)行器會(huì)通知調(diào)度程序。

內(nèi)置的三種Trigger觸發(fā)器類(lèi)型：

date:特定時(shí)間僅運(yùn)行一次作業(yè) interval: 固定的時(shí)間間隔內(nèi)運(yùn)行一次作業(yè) cron: 在一天內(nèi)特定的時(shí)間定期運(yùn)行作業(yè)

常見(jiàn)的Scheduler調(diào)度器:

BlockingScheduler: 調(diào)度程序是流程中唯一運(yùn)行的東西 BackgroundScheduler: 調(diào)度程序在應(yīng)用程序內(nèi)部的后臺(tái)運(yùn)行時(shí)使用 AsyncIOScheduler: 應(yīng)用程序使用asyncio模塊 GeventScheduler: 應(yīng)用程序使用gevent模塊 TornadoScheduler：構(gòu)建Tornado應(yīng)用程序時(shí)使用 TwistedScheduler: 構(gòu)建Tornado應(yīng)用程序時(shí)使用 QtScheduler: 在構(gòu)建QT應(yīng)用程序時(shí)使用

常見(jiàn)的JobStore：

MemoryJobStore MongoDBJobStore SQLAlchemyJobStore RedisJobStore 進(jìn)階使用

通過(guò)上面一個(gè)簡(jiǎn)單的示例了解大概的工作流程，以及各個(gè)組件在整個(gè)流程中的作用，以下的示例是Flask Web框架結(jié)合使用Apscheduler定時(shí)器，定時(shí)執(zhí)行任務(wù)。

# -*- coding: utf-8 -*-from flask import Flask, Blueprint, requestfrom apscheduler.executors.pool import ThreadPoolExecutor from apscheduler.schedulers.background import BackgroundSchedulerfrom apscheduler.jobstores.redis import RedisJobStoreimport timeapp = Flask(__name__)executors = {'default': ThreadPoolExecutor(5)}default_redis_jobstore = RedisJobStore(db=2, jobs_key='apschedulers.default_jobs', run_times_key='apschedulers.default_run_times', host = ’127.0.0.1’, port = 6379 )scheduler = BackgroundScheduler(executors=executors)scheduler.add_jobstore(default_redis_jobstore)scheduler.start()def say_hello(): print(time.strftime('%c'))@app.route('/get_job', methods=[’GET’])def get_job(): if scheduler.get_job('say_hello_test'): return 'YES' else: return 'NO'@app.route('/start_job', methods=['GET'])def start_job(): if not scheduler.get_job('say_hello_test'): scheduler.add_job(say_hello, 'interval', seconds=5, id='say_hello_test') return 'Start Scuessfully!' else: return 'Started Failed' @app.route('/remove_job', methods=['GET'])def remove_job(): if scheduler.get_job('say_hello_test'): scheduler.remove_job('say_hello_test') return 'Delete Successfully!' else: return 'Delete Failed'if __name__ == '__main__': app.run(host='127.0.0.1', port=8787, debug=True) 先分析Jobstore，這里使用的是RedisJobstore，將任務(wù)序列化存入到Redis數(shù)據(jù)庫(kù)中。這里順便提一下，為什么需要設(shè)置作業(yè)存儲(chǔ)器，原因是當(dāng)調(diào)度器程序崩潰時(shí)，仍然能夠保留作業(yè)，當(dāng)然選擇什么作業(yè)存儲(chǔ)器，可以根據(jù)具體的工作場(chǎng)景，目前主流的mysql，mongodb，redis，SQLite基本都支持; 然后再看看Scheduler，這里使用的時(shí)BackgroundScheduler，因?yàn)檫@里要求調(diào)度程序不能阻塞flask程序的正常接收請(qǐng)求，所以選在BackgrounScheduler讓它在開(kāi)始執(zhí)行任務(wù)時(shí)是在后臺(tái)運(yùn)行的，不會(huì)阻塞主線(xiàn)程; 最后看看工作的邏輯，這里get_job獲取作業(yè)的狀態(tài)，查看作業(yè)是否存在，start_job則是先判斷作業(yè)是否啟動(dòng)，然后再?zèng)Q定啟動(dòng)操作，remove_job則是停止作業(yè)。而這里的作業(yè)定義則是通過(guò)interval觸發(fā)器，每五秒執(zhí)行一次say_hello任務(wù)；總結(jié)

最后總結(jié)一下，首先你要設(shè)置一個(gè)作業(yè)存儲(chǔ)器用于在調(diào)度程序崩潰重新恢復(fù)時(shí)，還能夠在作業(yè)存儲(chǔ)器中獲取到作業(yè)繼續(xù)執(zhí)行；然后你需要設(shè)置一個(gè)執(zhí)行器，這個(gè)根據(jù)作業(yè)的類(lèi)型，比如時(shí)一個(gè)CPU密集型的任務(wù)，那就可以用進(jìn)程池執(zhí)行器，默認(rèn)是用線(xiàn)程池執(zhí)行器；最后創(chuàng)建配置調(diào)度器，啟動(dòng)調(diào)度，可以在啟動(dòng)前添加作業(yè)，也可以在啟動(dòng)后添加，刪除，獲取作業(yè)。(在這里需要明白的一點(diǎn)就是應(yīng)用程序不會(huì)直接去操作作業(yè)存儲(chǔ)器，作業(yè)或者執(zhí)行器，而是調(diào)度器提供適當(dāng)?shù)慕涌趤?lái)處理這些接口。)

ApScheduler是一個(gè)不錯(cuò)的定時(shí)任務(wù)庫(kù)，能夠動(dòng)態(tài)的添加刪除，同時(shí)也支持不同的觸發(fā)器類(lèi)型，這也是它的優(yōu)勢(shì)，相反一些如果是靜態(tài)任務(wù)，其實(shí)可以用如linux的crontab工具去做定時(shí)任務(wù)。有關(guān)這方面的記錄還會(huì)持續(xù)更新，如果有什么問(wèn)題，可以提出來(lái)，大家一起探討。

以上就是python Apscheduler的使用方法的詳細(xì)內(nèi)容，更多關(guān)于python Apscheduler的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！