APScheduler任務調度利器

Flink_China 發布于2019-07-25 11:11 / 2899人閱讀

摘要：中任務調度一般用中的任務調度工具也有不少等。調度器配置示例方式一方式二三略。移除調用放到，參數為調用實例的方法注意如果任務已經調度完畢，并且之后也不會再被執行的情況下，會被自動移除。可以監聽調度任務執行情況相關的事件。

Java中任務調度一般用Quartz,Python中的任務調度工具也有不少:Celery,RQ,APScheduler等。
Celery：非常強大的分布式任務調度框架
RQ：基于Redis的作業隊列工具
APScheduler：一款強大的任務調度工具

RQ參考Celery，據說要比Celery輕量級(Really?)
APScheduler感覺更像Quartz。
本人小小的建議是一般項目用APScheduler，因為不用像Celery那樣再多帶帶啟動worker、beat進程，而且API也很簡潔。
對于大點分布式項目用Celery

官網：http://apscheduler.readthedoc...
API：http://apscheduler.readthedoc...
當前版本：3.3.0
安裝：$ pip install apscheduler
例子：https://github.com/agronholm/...

特性

Advanced Python Scheduler (APScheduler) 一款強大的任務調度工具.
內置了三種調度模式：

Cron風格

間隔性(Interval-based)執行

僅在某個時間執行一次

作業存儲支持以下幾種方式：
Memory
SQLAlchemy (any RDBMS supported by SQLAlchemy works)
MongoDB
Redis
RethinkDB
ZooKeeper
除了Memory方式不需要序列化之外(一個例外是使用ProcessPoolExecutor)，其余都需要作業函數參數可序列化。

支持與以下框架集成:
asyncio (PEP 3156)
gevent
Tornado
Twisted
Qt (using either PyQt or PySide)

基本概念

四大組件：
triggers
job stores
executors
schedulers

schedulers

內置以下幾種調度器實現：
BlockingScheduler:
BackgroundScheduler: 默認采用ThreadPoolExecutor池(默認10)，可以配置ProcessPoolExecutor，或同時使用
AsyncIOScheduler: 使用asyncio模塊
GeventScheduler: 使用gevent
TornadoScheduler: use if you’re building a Tornado application
TwistedScheduler: use if you’re building a Twisted application
QtScheduler: use if you’re building a Qt application

基類：BaseScheduler，可以通過此類查詢相關配置選項。

調度器配置示例：

方式一、

from pytz import utc

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

jobstores = {
    "mongo": MongoDBJobStore(),
    "default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")
}
executors = {
    "default": ThreadPoolExecutor(20),
    "processpool": ProcessPoolExecutor(5)
}
job_defaults = {
    "coalesce": False,
    "max_instances": 3
}
scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

方式二、三：略。請看官網

啟動與關閉調度器：

通過調用start()方法來啟動

scheduler.start()
scheduler.shutdown()
scheduler.shutdown(wait=False)

triggers

內置以下三種觸發器實現：
date: 對已僅執行一次的情緒，指定某個之間點。請參考這里
interval: 指定時間間隔(fixed intervals)周期性執行。請參考這里
cron: 使用cron風格表達式周期性執行。請參考這里

job管理

添加jobs:

調用scheduler.add_job()方法，會返回apscheduler.job.Job實例(可用于job修改、移除等)

使用裝飾器scheduled_job()

作業存儲注意事項：

除了Memory方式不需要序列化之外(一個例外是使用ProcessPoolExecutor)，其余都需要作業函數參數可序列化。
如果需要存儲作業，而且每次啟動時你的應用都會重新添加一遍作業，那么請在添加job時指定一個唯一的ID，以及指定replace_existing=True，否則每次啟動應用都會添加一次job的副本。
如果需要立即啟動該任務，請在添加job時指定trigger參數。

移除job:
調用scheduler.remove_job()放到，參數為 job’s ID and job store alias
調用job實例的remove()方法 on the Job instance you got from add_job()

注意：如果任務已經調度完畢，并且之后也不會再被執行的情況下，會被自動移除。

job = scheduler.add_job(myfunc, "interval", minutes=2)
job.remove()

scheduler.add_job(myfunc, "interval", minutes=2, id="my_job_id")
scheduler.remove_job("my_job_id")

暫停和恢復job：

apscheduler.job.Job.pause()
apscheduler.schedulers.base.BaseScheduler.pause_job()

apscheduler.job.Job.resume()
apscheduler.schedulers.base.BaseScheduler.resume_job()

獲取jobs列表

apscheduler.get_jobs()

修改job：

可以通過apscheduler.job.Job.modify() or apscheduler.modify_job()修改除了id之外的job屬性。

job.modify(max_instances=6, name="Alternate name")

如果你想修改job的調度器，你可以使用apscheduler.job.Job.reschedule() or reschedule_job()

scheduler.reschedule_job("my_job_id", trigger="cron", minute="*/5")

限制同一個job實例的并發執行數

默認情況下同一個job，只允許一個job實例運行。這在某個job在下次運行時間到達之后仍未執行完畢時，能達到控制的目的。你也可以該變這一行為，在你調用add_job時可以傳遞max_instances=5來運行同時運行同一個job的5個job實例。

job錯過執行時間與job合并(Missed job executions and coalescing):

一個job可能由于某些情況錯過執行時間,比如上一點提到的，或者是線程池或進程池用光了，或者是當要調度job時，突然down機了等。
這時可以通過設置job的misfire_grace_time選項來指示之后嘗試執行的次數。
當然如果這不符合你的期望，你可以合并所有錯過時間的job到一個job來執行，通過設定job的coalesce=True。

Scheduler events

可以監聽調度、任務執行情況相關的事件。

def my_listener(event):
    if event.exception:
        print("The job crashed :(")
    else:
        print("The job worked :)")

scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

支持的事件列表：
http://apscheduler.readthedoc...

小結

有木有非常強大，又非常易于理解的感覺。Good Job！

GPU云服務器云服務器任務調度任務調度過程任務調度算法任務調度開源

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/38258.html

Python任務調度模塊APScheduler

介紹官網文檔：http://apscheduler.readthedoc...API：http://apscheduler.readthedoc... APScheduler是一個python的第三方庫，用來提供python的后臺程序。包含四個組件，分別是： triggers：任務觸發器組件，提供任務觸發方式 job stores：任務商店組件，提供任務保存方式 executors：任務...

zxhaaa 2019-07-30 14:38 評論0 收藏0
定時任務框架APScheduler學習詳解

摘要：安裝利用進行安裝源碼安裝有四種組成部分觸發器包含調度邏輯，每一個作業有它自己的觸發器，用于決定接下來哪一個作業會運行。除了他們自己初始配置意外，觸發器完全是無狀態的。 APScheduler簡介在平常的工作中幾乎有一半的功能模塊都需要定時任務來推動，例如項目中有一個定時統計程序，定時爬出網站的URL程序，定時檢測釣魚網站的程序等等，都涉及到了關于定時任務的問題，第一時間想到的是利用t...

sewerganger 2019-07-31 11:19 評論0 收藏0
Python Apscheduler源代碼解析(一) 任務調度流程

摘要：最近公司有項目需要使用到定時任務，其定時邏輯類似于的，就使用了這個類庫。在一次循環結束之前會計算任務下次執行事件與當前時間之差，然后讓調度線程掛起直到那個時間到來。最近公司有項目需要使用到定時任務，其定時邏輯類似于linux的Cron，就使用了Apscheduler這個類庫。基于公司的業務，需要修改Apshceduler，故而研究了一下Apscheduler的代碼。 Apschedu...

chavesgu 2019-07-30 14:53 評論0 收藏0
Flask-APScheduler使用教程

摘要：項目中需要用到定時器和循環執行。運用線程執行輪詢操作，也有運用系統的的文章最多，但是太麻煩。和中間人的消息傳輸支持所有特性，但也提供大量其他實驗性方案的支持，包括用進行本地開發。同時也包含了對任務的控制。后續有需求在繼續。項目中需要用到定時器和循環執行。去網上搜了一下，比較常見的有一下集中。運用Python線程執行輪詢操作，也有運用Linux系統的Cron,Celery的文章最多，但...

Noodles 2019-07-30 17:49 評論0 收藏0
關于Flask Schedule

摘要：日期觸發一次性指定日期作業的運行日期或時間指定時區運行一次運行一次間隔調度間隔幾周間隔幾天間隔幾小時間隔幾分鐘間隔多少秒開始日期結束日期時區每兩個小時調一下觸發年，位數字月范圍日范圍周范圍周內第幾天或者星期幾范圍或者時范圍 Flask Schedule Flask-APScheduler a Flask extension supported for the APScheduler w...

鄒強 2019-07-31 10:36 評論0 收藏0