python任务调度框架_python任务调度框架apscheduler【转】
簡介
APScheduler(以下簡稱APS)框架可以讓用戶定時執行或者周期性執行Python任務。既可以添加任務也可以刪除任務，還可以將任務存儲在數據庫中。當APS重啟之后，還會繼續執行之前設置的任務。
APS是跨平臺的，注意APS既不是守護進程也不是服務，更不是命令行程序。APS是進程內的調度器，也就是說它的實現原理是在進程內產生內置的阻塞來創建定時服務，以便在預定的時間內執行某個任務。
APS支持以下三種定時任務：
crontab類型任務
固定時間間隔任務
基于日期時間的一次性任務
你可以使用上面任意一種定時任務或者組合任務。
APS支持以下面幾種方式保存：
內存
SQLAlchemy
MongoDB
Redis
APS也可以整合進其他幾個Python框架：
asyncio
gevent
Tornado
Twisted
Qt
安裝
使用pip進行安裝：
$ pip install apscheduler
如果你沒有安裝pip，可以通過下面鏈接進行安裝get-pip.py
如果不能通過pip下載，也可以直接下載APS安裝包，解壓再進行安裝
$ python setup.py install
基礎概念
APS由以下幾部分組成：
觸發器(triggers)
任務倉庫(job stores)
執行器(executors)
調度器(schedulers)
觸發器包含了所有定時任務邏輯，每個任務都有一個對應的觸發器，觸發器決定任何的何時執行，初始配置情況下，觸發器是無狀態的。
任務倉庫保存要執行的任務，其中一個默認的任務倉庫將任務保存在內存中，而另外幾個任務倉庫將任務保存在數據庫中。在將任務保存到任務倉庫前，會對任務執行序列化操作，當重新讀取任務時，再執行反序列化操作。除了默認的任務倉庫，其他任務倉庫都不會在內存中保存任務，而是作為任務保存、加載、更新以及搜索的一個中間件。任務倉庫在定時器之間不能共享。
執行器用來執行定時任務，它只是將要執行的任務放在新的線程或者線程池中運行。執行完畢之后，再通知定時器。
調度器將其它幾個組件聯系在一起，一般在應用中只有一個調度器，程序開發者不會直接操作觸發器、任務倉庫或執行器，相反，調度器提供了這個接口。任務倉庫以及執行器的配置都是通過調度器來實現的。
選擇合適的調度器、執行器以及任務倉庫
選擇調度器是根據我們的開發環境與實際應用來決定的，下面是一些常用調度器：
BlockingScheduler：適合于只在進程中運行單個任務的情況
BackgroundScheduler: 適合于要求任何在程序后臺運行的情況
AsyncIOScheduler：適合于使用asyncio框架的情況
GeventScheduler: 適合于使用gevent框架的情況
TornadoScheduler: 適合于使用Tornado框架的應用
TwistedScheduler: 適合使用Twisted框架的應用
QtScheduler: 適合使用QT的情況
而任務倉庫，如果是非持久任務，使用MemoryStore就可以了，如果是持久性任務，那么久需要根據編程環境進行選擇了。
大多數情況下，執行器選擇ThreadPoolExecutor就可以了，但是如果涉及到比較耗CPU的任務，就可以選擇ProcessPoolExecutor，以充分利用多核CPU。，當然也可以同時使用兩個執行器。
可以在相應的API中找到對應的任務倉庫以及執行器。
配置任務調度器
APS提供了多種調度器配置方法，既可以使用配置字典，也可以直接傳遞配置參數給調度器，還可以先初始化調度器，添加完任務之后，最后再來配置調度器。
相關的配置參數可以參考API文檔。
舉個簡單例子：
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
# Initialize the rest of the application here, or before the scheduler initialization
上面的代碼生成一個后臺調度器，使用默認名為default的MemoryJobStore，以及默認名為default的ThreadPoolExecutor，最大線程數為10。
現在假設你想使用兩個任務倉庫以及兩個執行器，并且還想調整下任務的默認參數。可以使用下面三種方法，所完成的功能是一樣的。
方法1：
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)
方法2：
from apscheduler.schedulers.background import BackgroundScheduler
# The "apscheduler." prefix is hard coded
scheduler = BackgroundScheduler({
'apscheduler.jobstores.mongo': {
'type': 'mongodb'
},
'apscheduler.jobstores.default': {
'type': 'sqlalchemy',
'url': 'sqlite:///jobs.sqlite'
},
'apscheduler.executors.default': {
'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
'max_workers': '20'
},
'apscheduler.executors.processpool': {
'type': 'processpool',
'max_workers': '5'
},
'apscheduler.job_defaults.coalesce': 'false',
'apscheduler.job_defaults.max_instances': '3',
'apscheduler.timezone': 'UTC',
})
方法3：
from pytz import utc
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor
jobstores = {
'mongo': {'type': 'mongodb'},
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
'default': {'type': 'threadpool', 'max_workers': 20},
'processpool': ProcessPoolExecutor(max_workers=5)
}
job_defaults = {
'coalesce': False,
'max_instances': 3
}
scheduler = BackgroundScheduler()
# .. do something else here, maybe add jobs etc.
scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
啟動調度器
使用start()方法啟動調度器，BlockingScheduler需要在初始化之后才能執行start(),對于其他的Scheduler，調用start()方法都會直接返回，然后可以繼續執行后面的初始化操作。
調度器啟動之后，就不能更改它的配置了。
添加任務
有兩種添加任務的辦法：
調用add_job()
使用scheduled_job()修飾器
第一個方法是使用最多的，因為調用它會返回一個apscheduler.job.Job實例，后續可以對它進行修改或者刪除，而使用修飾器添加的任務添加之后就不能進行修改。
在調度器中設置定時任務，如果任務添加的時候，調度器還沒有啟動，那么任務只是暫時放到調度器中，當調度器啟動之后重新計算第一次執行時間。
需要注意的是，可以使用執行器或者任務倉庫來序列化任務，但是這個任務必須滿足兩個條件：
調用對象必須全局可訪問
調用對象的參數必須也可以序列化。
在所有內置的任務倉庫中，只有MemoryJobStore不能序列化對象；在所有內置的執行器中，只有ProcessPoolExecutors會序列化任務。
如果你需要在應用中使用持久性任務，那就必須給任務定義一個ID，并設置replace_existing=True,否則每次重啟應用都會返回一個新的任務。
如果要立即執行任務，只需要在添加任務的時候省略trigger參數。
刪除任務
當從調度器中刪除任務的時候，就會從相關聯的任務倉庫中刪除任務，后面就不會再執行了，有兩種方法來刪除任務。
調用remove_job()，參數為job ID以及任務倉庫名
調用remove()
第二種方法用起來更加方便，但是需要先保存添加任務時返回的實例對象；而通過scheduled_job()添加的任務，只能使用第一種方法進行刪除。
如果任務執行完畢，它會自動被刪除。
例如：
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')
暫停與恢復任務
暫停與恢復任務也很簡單，可以直接操作任務實例或者調度器來實現，當任務暫停時，它的運行時間會被重置，暫停期間不會計算進去，重啟之后又算進去。
暫停任務可以使用以下兩種方法：
apscheduler.job.Job.pause()
apscheduler.schedulers.base.BaseScheduler.pause_job()
恢復任務可以使用以下兩種方法：
apscheduler.job.Job.resume()
apscheduler.schedulers.BaseScheduler.resume_job()
獲取任務列表
可以使用get_jobs()方法來獲取當前正在處理的任務列表，如果只是想獲取某個任務倉庫中的任務列表，可以使用任務倉庫名作為參數傳入。
APS還提供了一個print_jobs()方法來打印格式化的任務列表。
修改任務
使用apscheduler.job.Job.modify()或者modify_job()方法可以修改任務的屬性，你可以修改任務的任意屬性，除了id。
例如：
job.modify(max_instances=6, name='Alternate name')
如果你想重新調度某個任務，例如改變它的觸發器，則可以使用apscheduler.job.Job.reschedule()或者reschedule_job()方法，這個兩個方法都可以為任務重新創建一個觸發器，并重新計算任務的運行時間。
例如：
scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')
關閉調度器
使用下面的方法關閉調度器：
scheduler.shutdown()
默認情況下，scheduler會關閉它的任務倉庫以及執行器，并等待所有正在執行的任務執行完。如果你不想等待，可以這樣操作：
scheduler.shutdown(wait=False)
限制任務實例并發執行的個數
默認情況下，只允許同時執行一個任務實例，通過max_instances參數來設置允許并發執行任務的個數。
錯失任務的合并
有時候調度器可能無法按時執行某個任務，最常見的情況就是當某個持久性任務保存在任務倉庫中時，調度器關閉之后再重啟，但是任務需要在重啟之前被執行，這時這個任務就被錯過了。調度器會根據任務misfire_grace_time參數的配置來決定錯過的任務還要不要繼續執行。這樣做的話可能導致重啟調度器之后會連續執行好幾個任務。
調度器事件
可以給調度器添加事件監聽器，調度器事件只有在某些情況下才會被觸發，并且可以攜帶某些有用的信息。通過給add_listener()傳遞合適的mask參數，可以只監聽幾種特定的事件類型。
例如：
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)
總結
以上是生活随笔為你收集整理的python任务调度框架_python任务调度框架apscheduler【转】的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： python爬虫背景_利用Python代
- 下一篇： websocket python爬虫_p