APScheduler（Python化的Cron）使用總結 定時任務

APScheduler（Python化的Cron）使用總結

簡介

APScheduler全程爲Advanced Python Scheduler，是一款輕量級的Python任務調度框架。它容許你像Cron那樣安排按期執行的任務，而且支持Python函數或任意可調用的對象。官方文檔：https://apscheduler.readthedocs.io/en/latest/userguide.html#basic-concepts

APScheduler安裝

方法一：使用pip安裝

$ pip install apscheduler 
方法二：若是pip不起做用，能夠從pypi上下載最新的源碼包(https://pypi.python.org/pypi/APScheduler/)進行安裝：

$ python setup.py install 
APScheduler組件

triggers（觸發器）： 觸發器中包含調度邏輯，每一個做業都由本身的觸發器來決定下次運行時間。除了他們本身初始配置意外，觸發器徹底是無狀態的。

job stores（做業存儲器）：存儲被調度的做業，默認的做業存儲器只是簡單地把做業保存在內存中，其餘的做業存儲器則是將做業保存在數據庫中。看成業被保存到一個持久化的做業存儲器中的時候，該做業的數據會被序列化，並在加載時被反序列化。做業存儲器不能共享調度器。

executors（執行器）：處理做業的運行，他們一般經過在做業中提交指定的可調用對象到一個線程或者進城池來進行。看成業完成時，執行器將會通知調度器。

schedulers（調度器）：配置做業存儲器和執行器能夠在調度器中完成，例如添加、修改和移除做業。根據不一樣的應用場景能夠選用不一樣的調度器，可選的有BlockingScheduler,BackgroundScheduler,AsyncIOScheduler,GeventScheduler,TornadoScheduler,TwistedScheduler,QtScheduler 7種。

調度器

BlockingScheduler : 當調度器是你應用中惟一要運行的東西時。 
BackgroundScheduler : 當你沒有運行任何其餘框架並但願調度器在你應用的後臺執行時使用（充電樁即便用此種方式）。 
AsyncIOScheduler : 當你的程序使用了asyncio（一個異步框架）的時候使用。 
GeventScheduler : 當你的程序使用了gevent（高性能的Python併發框架）的時候使用。 
TornadoScheduler : 當你的程序基於Tornado（一個web框架）的時候使用。 
TwistedScheduler : 當你的程序使用了Twisted（一個異步框架）的時候使用 
QtScheduler : 若是你的應用是一個Qt應用的時候可使用。 
做業存儲器

若是你的應用在每次啓動的時候都會從新建立做業，那麼使用默認的做業存儲器（MemoryJobStore）便可，可是若是你須要在調度器重啓或者應用程序奔潰的狀況下任然保留做業，你應該根據你的應用環境來選擇具體的做業存儲器。例如：使用Mongo或者SQLAlchemy JobStore （用於支持大多數RDBMS）

執行器

對執行器的選擇取決於你使用上面哪些框架，大多數狀況下，使用默認的ThreadPoolExecutor已經可以知足需求。若是你的應用涉及到CPU密集型操做，你能夠考慮使用ProcessPoolExecutor來使用更多的CPU核心。你也能夠同時使用二者，將ProcessPoolExecutor做爲第二執行器。

觸發器

當你調度做業的時候，你須要爲這個做業選擇一個觸發器，用來描述這個做業什麼時候被觸發，APScheduler有三種內置的觸發器類型:

date 一次性指定日期 
interval 在某個時間範圍內間隔多長時間執行一次 
cron 和Linux crontab格式兼容，最爲強大 
date

最基本的一種調度，做業只會執行一次。它的參數以下：

run_date (datetime|str) – 做業的運行日期或時間 
timezone (datetime.tzinfo|str) – 指定時區 
舉個栗子：

# 2016-12-12運行一次job_function
sched.add_job(job_function, 'date', run_date=date(2016, 12, 12), args=['text'])
# 2016-12-12 12:00:00運行一次job_function
sched.add_job(job_function, 'date', run_date=datetime(2016, 12, 12, 12, 0, 0), args=['text'])
interval

 

間隔調度，參數以下：

weeks (int) – 間隔幾周 
days (int) – 間隔幾天 
hours (int) – 間隔幾小時 
minutes (int) – 間隔幾分鐘 
seconds (int) – 間隔多少秒 
start_date (datetime|str) – 開始日期 
end_date (datetime|str) – 結束日期 
timezone (datetime.tzinfo|str) – 時區 
舉個栗子：

 

# 每兩個小時調一下job_function
sched.add_job(job_function, 'interval', hours=2)
cron

 

參數以下：

year (int|str) – 年，4位數字 
month (int|str) – 月 (範圍1-12) 
day (int|str) – 日 (範圍1-31) 
week (int|str) – 周 (範圍1-53) 
day_of_week (int|str) – 周內第幾天或者星期幾 (範圍0-6 或者 mon,tue,wed,thu,fri,sat,sun) 
hour (int|str) – 時 (範圍0-23) 
minute (int|str) – 分 (範圍0-59) 
second (int|str) – 秒 (範圍0-59) 
start_date (datetime|str) – 最先開始日期(包含) 
end_date (datetime|str) – 最晚結束時間(包含) 
timezone (datetime.tzinfo|str) – 指定時區 
舉個栗子：

 

# job_function將會在6,7,8,11,12月的第3個週五的1,2,3點運行
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
# 截止到2016-12-30 00:00:00，每週一到週五早上五點半運行job_function
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2016-12-31')

 

配置調度程序

在應用程序中使用默認做業存儲和默認執行程序運行BackgroundScheduler的例子：

 

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()
    # Initialize the rest of the application here, or before the scheduler initialization

 

這將生成一個名爲「default」的MemoryJobStore和名爲「default」的ThreadPoolExecutor的BackgroundScheduler，默認最大線程數爲10。

若是不知足於當前配置，如但願使用兩個執行器有兩個做業存儲器，而且還想要調整新做業的默認值並設置不一樣的時區，可按以下配置：

 

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     # 設置調度程序將同時運行的特定做業的最大實例數3
}
scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

 

 

啓動調度器

啓動調度器只須要調用start()方法，對於除BlockingScheduler之外的調度程序，此調用將當即返回，您能夠繼續應用程序的初始化過程，可能會將做業添加到調度程序。

對於BlockingScheduler，只需在完成初始化步驟後調用start（）

 

scheduler.start()

 

添加做業

方法一：調用add_job()方法

最多見的方法，add_job（）方法返回一個apscheduler.job.Job實例，您能夠稍後使用它來修改或刪除該做業。

方法二：使用裝飾器scheduled_job()

此方法主要是方便的聲明在應用程序運行時不會改變的做業

刪除做業

方法一：經過做業ID或別名調用remove_job()刪除做業

方法二：經過add_job()返回的job實例調用remove()方法刪除做業

舉個栗子：

# 實例刪除
job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
 # id刪除
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()

 

獲取做業列表

要得到計劃做業的機器可處理列表，可使用get_jobs（）方法。 它將返回一個Job實例列表。 若是您只對特定做業存儲中包含的做業感興趣，則將做業存儲別名做爲第二個參數。

爲了方便起見，可使用print_jobs（）方法，它將打印格式化的做業列表，觸發器和下次運行時間。

修改做業屬性

您能夠經過調用apscheduler.job.Job.modify（）或modify_job（）來修改除id之外的任何做業屬性。

 

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

 

關閉調度程序

默認狀況下，調度程序關閉其做業存儲和執行程序，並等待全部當前正在執行的做業完成，wait=False參數可選，表明當即中止，不用等待。

scheduler.shutdown(wait=False) 
附：一、定時任務運行腳本小例子：

 

import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
from app.untils.log_builder import sys_logging

scheduler = BlockingScheduler()   # 後臺運行

 # 設置爲每日凌晨00:30:30時執行一次調度程序
@scheduler.scheduled_job("cron", day_of_week='*', hour='1', minute='30', second='30')
def rebate():
        print "schedule execute"
        sys_logging.debug("statistic scheduler execute success" + datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))


if __name__ == '__main__':
    try:
        scheduler.start()
        sys_logging.debug("statistic scheduler start success")
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        sys_logging.debug("statistic scheduler start-up fail")

 

二、第一次使用此定時器時總會執行兩次，一直不知道爲何，後來發現，python 的flask框架在debug模式下會多開一個線程監測項目變化，因此每次會跑兩遍，能夠將debug選項改成False