任務(wù)調(diào)度后端【免費(fèi)下載鏈接】apschedulerTask scheduling library for Python項(xiàng)目地址https://gitcode.com/gh_mirrors/ap/apscheduler點(diǎn)擊查看免費(fèi)下載APSchedulerAdvanced Python Scheduler是 Python 生態(tài)中兼具任務(wù)調(diào)度與任務(wù)隊(duì)列能力的庫既可用于單進(jìn)程內(nèi)的輕量定時任務(wù)也能通過共享數(shù)據(jù)存儲與事件代理擴(kuò)展到跨多節(jié)點(diǎn)的分布式部署。本篇以倉庫 README.rst 為骨架結(jié)合源碼實(shí)現(xiàn)系統(tǒng)講解其同步/異步雙形態(tài)、四大內(nèi)置觸發(fā)機(jī)制、持久化數(shù)據(jù)存儲與事件代理、并發(fā)上限、misfire 容錯與抖動等核心特性幫助讀者掌握從單進(jìn)程示例到多節(jié)點(diǎn)高可用的完整落地路徑。一、項(xiàng)目定位任務(wù)調(diào)度器與任務(wù)隊(duì)列的合體根據(jù) README.rstAPScheduler 首先是一個任務(wù)調(diào)度器 任務(wù)隊(duì)列系統(tǒng)。即使你完全沒有調(diào)度需求也可以只把它當(dāng)作純作業(yè)隊(duì)列來用——這意味著任務(wù)的執(zhí)行與觸發(fā)被解耦你可以只排隊(duì)、只執(zhí)行而不用關(guān)心何時觸發(fā)。它的伸縮性覆蓋兩個極端單進(jìn)程簡單場景一個進(jìn)程內(nèi)運(yùn)行調(diào)度器即可滿足開箱即用見 examples/standalone/sync_memory.py。多節(jié)點(diǎn)大規(guī)模部署多個調(diào)度器與 worker 共享同一個數(shù)據(jù)存儲從而同時獲得高可用HA與水平擴(kuò)展horizontal scaling能力。這種共享設(shè)計背后是調(diào)度與執(zhí)行分離的架構(gòu)任務(wù)、調(diào)度計劃Schedule與作業(yè)Job都持久化在共享數(shù)據(jù)存儲中任何一個節(jié)點(diǎn)都可以接替另一個節(jié)點(diǎn)的工作。同步與異步兩種形態(tài)APScheduler 同時提供同步與異步兩種使用形態(tài)覆蓋傳統(tǒng)線程型應(yīng)用和 asyncio/Trio 異步應(yīng)用同步形態(tài)Scheduler位于 src/apscheduler/_schedulers/sync.py。從源碼看它是一個圍繞AsyncScheduler的同步包裝器啟動時在一個獨(dú)立線程中運(yùn)行異步事件循環(huán)并在調(diào)度器關(guān)閉時一并關(guān)閉該線程。也就是說同步版本底層仍是異步事件循環(huán)只是對調(diào)用方隱藏了細(xì)節(jié)。異步形態(tài)AsyncScheduler位于 src/apscheduler/schedulers/async.py基于 AnyIO 實(shí)現(xiàn)需要 asyncio 或 Trio 作為底層事件循環(huán)。其文檔注釋還特別提示如果運(yùn)行在 Trio 上要確保數(shù)據(jù)存儲與事件代理與 Trio 兼容。兩者統(tǒng)一從 src/apscheduler/init.py 對外導(dǎo)出用戶只需from apscheduler import Scheduler或from apscheduler import AsyncScheduler即可。與此對應(yīng)examples/standalone 目錄中同時提供了sync_memory.py與async_memory.py兩個最小示例展示了同一套 API 在同步與異步環(huán)境下的用法。Web 應(yīng)用集成README 聲明提供了與 WSGI 或 ASGI 兼容 Web 應(yīng)用集成的文檔與示例。倉庫 examples/web 目錄中給出了五種集成示例WSGI 路線wsgi_flask.pyFlask 框架、wsgi_noframework.py無框架原生 WSGIASGI 路線asgi_fastapi.pyFastAPI、asgi_starlette.pyStarlette、asgi_noframework.py無框架原生 ASGI二、持久化數(shù)據(jù)存儲調(diào)度與作業(yè)的共享底座README 明確支持調(diào)度計劃與作業(yè)的持久化存儲這意味著它們可以在多個調(diào)度器/worker 實(shí)例間共享即使進(jìn)程或節(jié)點(diǎn)重啟調(diào)度計劃與作業(yè)依然存在。內(nèi)置數(shù)據(jù)存儲后端內(nèi)置的持久化數(shù)據(jù)存儲后端共四類見 README.rst后端倉庫實(shí)現(xiàn)PostgreSQL由 SQLAlchemy 數(shù)據(jù)存儲統(tǒng)一覆蓋MySQL 及其衍生品由 SQLAlchemy 數(shù)據(jù)存儲統(tǒng)一覆蓋SQLite由 SQLAlchemy 數(shù)據(jù)存儲統(tǒng)一覆蓋MongoDBsrc/apscheduler/datastores/mongodb.py其中 PostgreSQL/MySQL/SQLite 三者均由 src/apscheduler/datastores/sqlalchemy.py 中的SQLAlchemyDataStore實(shí)現(xiàn)——SQLAlchemy 的方言機(jī)制天然支持這三種數(shù)據(jù)庫乃至更多。從源碼看該數(shù)據(jù)存儲在列類型層面做了大量兼容處理例如EmulatedTimestampTZ用 Unicode 字符串ISO 格式模擬帶時區(qū)的時間戳src/apscheduler/datastores/sqlalchemy.pyEmulatedInterval用 BigInteger 模擬時間間隔從而在缺乏原生類型的數(shù)據(jù)庫上保持行為一致。除持久化后端外還有基于內(nèi)存的 src/apscheduler/datastores/memory.pyMemoryDataStore適合單進(jìn)程、無需持久化的場景——它也是AsyncScheduler的默認(rèn)數(shù)據(jù)存儲見 src/apscheduler/schedulers/async.py。數(shù)據(jù)存儲的用法示例在 examples/separate_worker/sync_scheduler.py 中可以看到真實(shí)用法用create_async_engine(postgresqlasyncpg://...)創(chuàng)建異步引擎然后傳給SQLAlchemyDataStore(engine)即可作為調(diào)度器的data_store參數(shù)。README 也提到schedule 與 job 會持久化并跨節(jié)點(diǎn)存活。from sqlalchemy.ext.asyncio import create_async_engine from apscheduler import Scheduler from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore from apscheduler.eventbrokers.asyncpg import AsyncpgEventBroker from apscheduler.triggers.interval import IntervalTrigger engine create_async_engine(postgresqlasyncpg://postgres:secretlocalhost/testdb) data_store SQLAlchemyDataStore(engine) event_broker AsyncpgEventBroker.from_async_sqla_engine(engine) with Scheduler(data_store, event_broker) as scheduler: scheduler.add_schedule(tick, IntervalTrigger(seconds1), idtick)注意該示例的核心思想是只寫調(diào)度、不啟動調(diào)度器——with塊內(nèi)沒有調(diào)用run_until_stopped()調(diào)度器僅作為寫入系統(tǒng)接口使用真正的執(zhí)行交由另一端的 worker 進(jìn)程完成見 examples/separate_worker/sync_worker.py 與async_worker.py。三、事件代理多節(jié)點(diǎn)協(xié)同的通知中樞當(dāng)存在多個調(diào)度器/worker 時需要事件代理event broker來廣播事件作業(yè)被添加、調(diào)度計劃被更新、作業(yè)被獲取/釋放等。內(nèi)置事件代理README 列出的內(nèi)置事件代理有三個README.rst事件代理倉庫實(shí)現(xiàn)依賴PostgreSQLsrc/apscheduler/eventbrokers/asyncpg.py、src/apscheduler/eventbrokers/psycopg.pyasyncpg / psycopgRedissrc/apscheduler/eventbrokers/redis.pyredisMQTTsrc/apscheduler/eventbrokers/mqtt.pypaho-mqtt此外還有進(jìn)程內(nèi)直接使用的 src/apscheduler/eventbrokers/local.pyLocalEventBroker它是AsyncScheduler的默認(rèn)事件代理見 src/apscheduler/schedulers/async.py適合單進(jìn)程場景。Redis 事件代理的源碼細(xì)節(jié)以RedisEventBroker為例src/apscheduler/eventbrokers/redis.py接受一個異步 Redis 客戶端或 Redis URL 字符串作為client_or_urlchannel參數(shù)指定消息廣播頻道默認(rèn)值為apschedulerstop_check_interval控制頻道監(jiān)聽器檢查是否需要停止的間隔單位秒值越大響應(yīng)停止越慢但 CPU 占用越低默認(rèn) 1 秒若傳入 URL 字符串代理會自行創(chuàng)建連接池并在退出時自動關(guān)閉_close_on_exit True若傳入現(xiàn)成客戶端實(shí)例代理不管理該客戶端的生命周期需由使用方自行關(guān)閉——這是官方注釋中明確提示的注意事項(xiàng)。啟動時start方法代理會訂閱指定頻道并啟動一個名為 Redis subscriber 的后臺任務(wù)持續(xù)監(jiān)聽消息src/apscheduler/eventbrokers/redis.py。所有外部代理都基于tenacity對連接失敗進(jìn)行自動重試見_retry方法及 src/apscheduler/_retry.py 中的RetrySettings。從數(shù)據(jù)存儲共享到事件廣播的完整鏈路多個調(diào)度器/worker 共享同一數(shù)據(jù)存儲因此任何節(jié)點(diǎn)都能看到全局的調(diào)度計劃與作業(yè)在此基礎(chǔ)上節(jié)點(diǎn)之間通過事件代理實(shí)時感知誰加了新調(diào)度、誰取走了作業(yè)、誰釋放了作業(yè)從而避免重復(fù)執(zhí)行并實(shí)現(xiàn)任務(wù)分配。這一模式對應(yīng) examples/separate_worker 目錄sync_scheduler.py/async_scheduler.py只負(fù)責(zé)登記調(diào)度sync_worker.py/async_worker.py負(fù)責(zé)實(shí)際執(zhí)行任務(wù)。四、內(nèi)置調(diào)度機(jī)制觸發(fā)器README 將內(nèi)置調(diào)度機(jī)制統(tǒng)稱為觸發(fā)器trigger共四種README.rst觸發(fā)器說明倉庫實(shí)現(xiàn)Cron 風(fēng)格調(diào)度仿照 UNIX cronsrc/apscheduler/triggers/cron/init.py間隔調(diào)度以固定間隔運(yùn)行src/apscheduler/triggers/interval.py日歷調(diào)度每隔 X 年/月/周/天觸發(fā)且總是在一天中的同一時刻src/apscheduler/triggers/calendarinterval.py一次性調(diào)度在指定日期/時間運(yùn)行一次src/apscheduler/triggers/date.pyCron 觸發(fā)器CronTrigger的構(gòu)造參數(shù)覆蓋year/month/day/week/day_of_week/hour/minute/second全套時間字段并提供start_time最早觸發(fā)時間默認(rèn)當(dāng)前時間、end_time最晚觸發(fā)時間與timezone時區(qū)默認(rèn)本地時區(qū)三個邊界控制參數(shù)src/apscheduler/triggers/cron/init.py。其內(nèi)部字段順序由FIELDS_MAP明確定義src/apscheduler/triggers/cron/init.py字段表達(dá)式解析邏輯在 src/apscheduler/triggers/cron/expressions.py 中實(shí)現(xiàn)。需要注意的語義細(xì)節(jié)一周的第一天固定是周一README 與源碼注釋一致強(qiáng)調(diào)day_of_week支持 0-7 的數(shù)字0 與 7 均為周日或sun,mon,...名稱支持標(biāo)準(zhǔn)的 crontab 五段表達(dá)式CronTrigger.from_crontab(expr)可把0 8 * * *這類表達(dá)式轉(zhuǎn)換為觸發(fā)器表達(dá)式必須恰好 5 段否則拋出ValueErrorsrc/apscheduler/triggers/cron/init.py。間隔觸發(fā)器IntervalTrigger接受weeks/days/hours/minutes/seconds/microseconds六個單位參數(shù)首次觸發(fā)時間為start_time創(chuàng)建時刻除非顯式指定若給出end_time最后一次觸發(fā)將不晚于該時刻src/apscheduler/triggers/interval.py。源碼中有兩個值得注意的約束src/apscheduler/triggers/interval.py總間隔必須為正數(shù)否則拋出ValueError(The time interval must be positive)end_time不得早于start_time否則拋出ValueError。最簡單的可用示例就是 examples/standalone/sync_memory.py 中的IntervalTrigger(seconds1)——每秒觸發(fā)一次tick()并打印當(dāng)前時間。日歷間隔觸發(fā)器CalendarIntervalTrigger實(shí)現(xiàn)每隔 X 年/月/周/天觸發(fā)、總是在一天中的同一時刻的日歷語義如每個月 1 號 09:00。它區(qū)別于IntervalTrigger的關(guān)鍵在于按日歷單位月、年長度不一計算而不是固定的時間增量。其實(shí)現(xiàn)位于 src/apscheduler/triggers/calendarinterval.py在 tests/triggers/test_calendarinterval.py 中有完整的測試覆蓋。一次性觸發(fā)器DateTrigger只需一個run_time參數(shù)僅在給定時刻觸發(fā)一次。源碼實(shí)現(xiàn)非常直白首次調(diào)用next()返回run_time并置_completed True此后一律返回Nonesrc/apscheduler/triggers/date.py。組合觸發(fā)器README 特別指出不同的調(diào)度機(jī)制還可以通過所謂的組合觸發(fā)器combining triggers結(jié)合使用。倉庫中 src/apscheduler/triggers/combining.py 提供了兩個組合器AndTrigger當(dāng)多個觸發(fā)器的觸發(fā)時刻落在給定閾值threshold默認(rèn) 1 秒內(nèi)時返回最早的那個觸發(fā)時刻否則讓最早觸發(fā)的子觸發(fā)器繼續(xù)前進(jìn)再重新比對并受max_iterations默認(rèn) 10000保護(hù)以防無限循環(huán)src/apscheduler/triggers/combining.py。OrTrigger返回任一子觸發(fā)器產(chǎn)生的下一個觸發(fā)時刻中最早的那個。組合觸發(fā)器的使用在 tests/triggers/test_combining.py 中有系統(tǒng)性的測試驗(yàn)證。自定義觸發(fā)器README 明確表示你可以通過構(gòu)建自己的觸發(fā)器類來實(shí)現(xiàn)自定義調(diào)度邏輯自定義觸發(fā)器與內(nèi)置觸發(fā)器地位完全平等不會被區(qū)別對待。這得益于抽象基類 src/apscheduler/abc.py 中定義的Trigger接口核心方法是next()返回下一個觸發(fā)時間。只要實(shí)現(xiàn)了該接口并處理好序列化所需的__getstate__/__setstate__如各內(nèi)置觸發(fā)器所示就可以自由接入調(diào)度器。五、其他值得注意的特性README 列舉了三項(xiàng)對生產(chǎn)環(huán)境極具價值的高級特性README.rst1. 限制同一任務(wù)的最大并發(fā)作業(yè)數(shù)可以為給定任務(wù)函數(shù)限制同時運(yùn)行的作業(yè)數(shù)量上限。源碼中對應(yīng)TaskDefaults.max_running_jobsAsyncScheduler在初始化時會為其設(shè)置默認(rèn)值 1src/apscheduler/schedulers/async.py即默認(rèn)情況下同一任務(wù)不會并發(fā)運(yùn)行——這避免了同一任務(wù)的重入競爭。2. 限制作業(yè)允許延遲啟動的時間可以為作業(yè)設(shè)置 misfire grace time誤觸發(fā)寬限時間限制一個作業(yè)被允許延遲多久才開始執(zhí)行。若實(shí)際觸發(fā)時間晚于計劃時間且超出寬限該次執(zhí)行將被放棄。源碼中對應(yīng)TaskDefaults.misfire_grace_time默認(rèn)值為None表示不限制src/apscheduler/schedulers/async.py同時配套CoalescePolicy/ConflictPolicy等枚舉見 src/apscheduler/_enums.py處理同一時刻多次觸發(fā)的合并策略與沖突策略。3. Jitter抖動可為每個計劃的作業(yè)運(yùn)行時間添加可調(diào)節(jié)的隨機(jī)延遲。這在多實(shí)例部署時尤其有用隨機(jī)抖動能打散各節(jié)點(diǎn)在同一時刻的觸發(fā)避免驚群效應(yīng)式的資源爭搶。此外從調(diào)度器構(gòu)造參數(shù)src/apscheduler/schedulers/async.py還可看到更多生產(chǎn)級配置項(xiàng)max_concurrent_jobs全局并發(fā)作業(yè)上限默認(rèn) 100、cleanup_interval自動清理過期數(shù)據(jù)的間隔默認(rèn) 15 分鐘、lease_duration對調(diào)度計劃/任務(wù)上鎖的最長持有時間默認(rèn) 30 秒、role指定調(diào)度器在運(yùn)行時只調(diào)度、只運(yùn)行作業(yè)、或兩者兼做等。六、版本狀態(tài)與生產(chǎn)使用前提README 開篇即給出重要警告README.rstv4.0 系列目前是預(yù)發(fā)布pre-release版本可能以不向后兼容的方式變化且沒有遷移路徑請勿在生產(chǎn)環(huán)境中使用該版本。與之對應(yīng)pyproject.toml 中聲明的核心依賴為anyio 4.0、attrs、tenacity、tzlocal并要求 Python 3.10分類器中列出了 3.10 至 3.14 的支持范圍。附加依賴按后端拆分asyncpg、cbor、mongodb、mqtt、redis、sqlalchemy等以可選依賴extras形式提供pyproject.toml用戶可按需安裝pip install APScheduler # 核心 pip install APScheduler[redis] # 啟用 Redis 事件代理 pip install APScheduler[mongodb] # 啟用 MongoDB 數(shù)據(jù)存儲 pip install APScheduler[sqlalchemy] # 啟用 SQLAlchemy 數(shù)據(jù)存儲 pip install APScheduler[mqtt] # 啟用 MQTT 事件代理七、快速上手同步內(nèi)存版最小示例把以上概念落到最小可運(yùn)行代碼上倉庫自帶的 examples/standalone/sync_memory.py 就是最直觀的入門樣例from datetime import datetime from apscheduler import Scheduler from apscheduler.triggers.interval import IntervalTrigger def tick(): print(Hello, the time is, datetime.now()) with Scheduler() as scheduler: scheduler.add_schedule(tick, IntervalTrigger(seconds1)) scheduler.run_until_stopped()運(yùn)行python sync_memory.py控制臺會以每秒一次的頻率打印當(dāng)前時間。這個例子雖然簡單卻完整展示了 APScheduler 的四個核心元素任務(wù)Task即tick函數(shù)由add_schedule自動注冊調(diào)度計劃Schedule任務(wù) 觸發(fā)器IntervalTrigger(seconds1)的綁定數(shù)據(jù)存儲默認(rèn)的MemoryDataStore事件代理默認(rèn)的LocalEventBroker。默認(rèn)情況下Scheduler()不傳參調(diào)度器內(nèi)部會自動裝配三個作業(yè)執(zhí)行器src/apscheduler/schedulers/async.pyasync異步執(zhí)行器、threadpool線程池執(zhí)行器、processpool進(jìn)程池執(zhí)行器對應(yīng) src/apscheduler/executors 目錄下的實(shí)現(xiàn)同步形態(tài)的Scheduler默認(rèn)任務(wù)執(zhí)行器為threadpoolsrc/apscheduler/_schedulers/sync.py。結(jié)語APScheduler 的設(shè)計在簡單與規(guī)模之間取得了平衡單進(jìn)程內(nèi)開箱即用的內(nèi)存數(shù)據(jù)存儲與本地事件代理與面向多節(jié)點(diǎn)的 PostgreSQL/MySQL/SQLite/MongoDB 持久化 PostgreSQL/Redis/MQTT 事件廣播共用同一套觸發(fā)器與調(diào)度 API。理解數(shù)據(jù)存儲與事件代理的職責(zé)邊界、四種觸發(fā)器及其組合/自定義機(jī)制、以及并發(fā)上限、misfire 寬限與 jitter 三項(xiàng)生產(chǎn)特性是正確使用并進(jìn)一步擴(kuò)展這套系統(tǒng)的關(guān)鍵。倉庫中的 examples 目錄提供了從獨(dú)立內(nèi)存版到分離 worker 版、再到 Web 集成版的完整進(jìn)階路線可作為動手實(shí)踐的參考。贊分享任務(wù)調(diào)度后端【免費(fèi)下載鏈接】apschedulerTask scheduling library for Python項(xiàng)目地址https://gitcode.com/gh_mirrors/ap/apscheduler點(diǎn)擊查看免費(fèi)下載相關(guān)推薦TDengine 3.x 分布式架構(gòu)深度解析邏輯節(jié)點(diǎn)、存儲模型與數(shù)據(jù)復(fù)制機(jī)制TDengine 3.x 分布式架構(gòu)深度解析邏輯節(jié)點(diǎn)、存儲模型與數(shù)據(jù)復(fù)制機(jī)制 TDengine 從誕生之初就采用分布式 高可用 水平擴(kuò)展的設(shè)計理念數(shù)據(jù)庫時序數(shù)據(jù)庫物聯(lián)網(wǎng)大數(shù)據(jù)實(shí)時分析云原生Sketch Constraints與原生約束對比哪款布局工具更適合你的設(shè)計需求Sketch Constraints與原生約束對比哪款布局工具更適合你的設(shè)計需求 在UI設(shè)計中高效的布局工具是提升工作效率的關(guān)鍵。Sketch作為主流設(shè)計CANN/asc-devkitSIMT-API half2對數(shù)函數(shù)h2log 產(chǎn)品支持情況 ! npu950 id1 Ascend 950PR/Ascend 950DT支持 ! end id1 ! npuA3人工智能深度學(xué)習(xí)算子庫CANNAscend上一篇如何通過Conventional Commits提升測試驅(qū)動開發(fā)效率5個實(shí)用技巧下一篇CANN/asc-devkit向量最大值標(biāo)量計算創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考