APScheduler任務(wù)調(diào)度利器

Flink_China 發(fā)布于2019-07-25 11:11 / 2906人閱讀

摘要：中任務(wù)調(diào)度一般用中的任務(wù)調(diào)度工具也有不少等。調(diào)度器配置示例方式一方式二三略。移除調(diào)用放到，參數(shù)為調(diào)用實(shí)例的方法注意如果任務(wù)已經(jīng)調(diào)度完畢，并且之后也不會(huì)再被執(zhí)行的情況下，會(huì)被自動(dòng)移除。可以監(jiān)聽(tīng)調(diào)度任務(wù)執(zhí)行情況相關(guān)的事件。

Java中任務(wù)調(diào)度一般用Quartz,Python中的任務(wù)調(diào)度工具也有不少:Celery,RQ,APScheduler等。
Celery：非常強(qiáng)大的分布式任務(wù)調(diào)度框架
RQ：基于Redis的作業(yè)隊(duì)列工具
APScheduler：一款強(qiáng)大的任務(wù)調(diào)度工具

RQ參考Celery，據(jù)說(shuō)要比Celery輕量級(jí)(Really?)
APScheduler感覺(jué)更像Quartz。
本人小小的建議是一般項(xiàng)目用APScheduler，因?yàn)椴挥孟馛elery那樣再多帶帶啟動(dòng)worker、beat進(jìn)程，而且API也很簡(jiǎn)潔。
對(duì)于大點(diǎn)分布式項(xiàng)目用Celery

官網(wǎng)：http://apscheduler.readthedoc...
API：http://apscheduler.readthedoc...
當(dāng)前版本：3.3.0
安裝：$ pip install apscheduler
例子：https://github.com/agronholm/...

特性

Advanced Python Scheduler (APScheduler) 一款強(qiáng)大的任務(wù)調(diào)度工具.
內(nèi)置了三種調(diào)度模式：

Cron風(fēng)格

間隔性(Interval-based)執(zhí)行

僅在某個(gè)時(shí)間執(zhí)行一次

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

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

基本概念

四大組件：
triggers
job stores
executors
schedulers

schedulers

內(nèi)置以下幾種調(diào)度器實(shí)現(xiàn)：
BlockingScheduler:
BackgroundScheduler: 默認(rèn)采用ThreadPoolExecutor池(默認(rèn)10)，可以配置ProcessPoolExecutor，或同時(shí)使用
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，可以通過(guò)此類查詢相關(guān)配置選項(xiàng)。

調(diào)度器配置示例：

方式一、

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)

方式二、三：略。請(qǐng)看官網(wǎng)

啟動(dòng)與關(guān)閉調(diào)度器：

通過(guò)調(diào)用start()方法來(lái)啟動(dòng)

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

triggers

內(nèi)置以下三種觸發(fā)器實(shí)現(xiàn)：
date: 對(duì)已僅執(zhí)行一次的情緒，指定某個(gè)之間點(diǎn)。請(qǐng)參考這里
interval: 指定時(shí)間間隔(fixed intervals)周期性執(zhí)行。請(qǐng)參考這里
cron: 使用cron風(fēng)格表達(dá)式周期性執(zhí)行。請(qǐng)參考這里

job管理

添加jobs:

調(diào)用scheduler.add_job()方法，會(huì)返回apscheduler.job.Job實(shí)例(可用于job修改、移除等)

使用裝飾器scheduled_job()

作業(yè)存儲(chǔ)注意事項(xiàng)：

除了Memory方式不需要序列化之外(一個(gè)例外是使用ProcessPoolExecutor)，其余都需要作業(yè)函數(shù)參數(shù)可序列化。
如果需要存儲(chǔ)作業(yè)，而且每次啟動(dòng)時(shí)你的應(yīng)用都會(huì)重新添加一遍作業(yè)，那么請(qǐng)?jiān)谔砑觠ob時(shí)指定一個(gè)唯一的ID，以及指定replace_existing=True，否則每次啟動(dòng)應(yīng)用都會(huì)添加一次job的副本。
如果需要立即啟動(dòng)該任務(wù)，請(qǐng)?jiān)谔砑觠ob時(shí)指定trigger參數(shù)。

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

注意：如果任務(wù)已經(jīng)調(diào)度完畢，并且之后也不會(huì)再被執(zhí)行的情況下，會(huì)被自動(dòng)移除。

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")

暫停和恢復(fù)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：

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

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

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

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

限制同一個(gè)job實(shí)例的并發(fā)執(zhí)行數(shù)

默認(rèn)情況下同一個(gè)job，只允許一個(gè)job實(shí)例運(yùn)行。這在某個(gè)job在下次運(yùn)行時(shí)間到達(dá)之后仍未執(zhí)行完畢時(shí)，能達(dá)到控制的目的。你也可以該變這一行為，在你調(diào)用add_job時(shí)可以傳遞max_instances=5來(lái)運(yùn)行同時(shí)運(yùn)行同一個(gè)job的5個(gè)job實(shí)例。

job錯(cuò)過(guò)執(zhí)行時(shí)間與job合并(Missed job executions and coalescing):

一個(gè)job可能由于某些情況錯(cuò)過(guò)執(zhí)行時(shí)間,比如上一點(diǎn)提到的，或者是線程池或進(jìn)程池用光了，或者是當(dāng)要調(diào)度job時(shí)，突然down機(jī)了等。
這時(shí)可以通過(guò)設(shè)置job的misfire_grace_time選項(xiàng)來(lái)指示之后嘗試執(zhí)行的次數(shù)。
當(dāng)然如果這不符合你的期望，你可以合并所有錯(cuò)過(guò)時(shí)間的job到一個(gè)job來(lái)執(zhí)行，通過(guò)設(shè)定job的coalesce=True。

Scheduler events

可以監(jiān)聽(tīng)調(diào)度、任務(wù)執(zhí)行情況相關(guān)的事件。

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...

小結(jié)

有木有非常強(qiáng)大，又非常易于理解的感覺(jué)。Good Job！