Python - APScheduler

目錄

• APScheduler簡介
  • 支持的後端存儲做業
  • 集成的Python框架
• APScheduler下載安裝
• APScheduler組件
  • 各組件簡介
    • 調度器
    • 做業存儲器
    • 執行器
    • 觸發器
• 使用
  • 添加做業
  • 只執行一次
  • 間隔執行

• 返回上一頁

APScheduler簡介

APScheduler（Advanced Python Scheduler）是一個輕量級的Python定時任務調度框架（Python庫）。
APScheduler有三個內置的調度系統，其中包括：

• cron式調度（可選開始/結束時間）
• 基於間隔的執行（以偶數間隔運行做業，也能夠選擇開始/結束時間）
• 一次性延遲執行任務（在指定的日期/時間內運行做業一次）

支持的後端存儲做業

APScheduler能夠任意混合和匹配調度系統和做業存儲的後端，其中支持後端存儲做業包括：

• Memory
• SQLAlchemy
• MongoDB
• Redis
• RethinkDB
• ZooKeeper

集成的Python框架

APScheduler內繼承了幾個常見的Python框架：

• asyncio
• gevent
• tornado
• qt

APScheduler下載安裝

使用pip安裝：

pip install apscheduler

若是超時或者出現別的狀況，能夠選擇：

# 法1使用豆瓣源下載
pip install -i https://pypi.doubanio.com/simple/ apscheduler
# 法2使用清華源下載
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple apscheduler

要是再不行，點擊該連接或者pypi官網下載了。下載並解壓縮，進入跟setup.py文件同級的目錄，打開cmd，使用命令進行下載：

python setup.py install

APScheduler組件

APScheduler共有4種組件，分別是：

• 觸發器（trigger），觸發器中包含調度邏輯，每一個做業都有本身的觸發器來決定下次運行時間。除了它們本身初始配置之外，觸發器徹底是無狀態的。
• 做業存儲器（job store），存儲被調度的做業，默認的做業存儲器只是簡單地把做業保存在內存中，其餘的做業存儲器則是將做業保存在數據庫中，看成業被保存在一個持久化的做業存儲器中的時候，該做業的數據會被序列化，並在加載時被反序列化，須要說明的是，做業存儲器不能共享調度器。
• 執行器（executor），處理做業的運行，一般經過在做業中提交指定的可調用對象到一個線程或者進程池來進行，看成業完成時，執行器會將通知調度器。
• 調度器（scheduler），配置做業存儲器和執行器能夠在調度器中完成。例如添加、修改、移除做業，根據不一樣的應用場景，能夠選擇不一樣的調度器，可選的將在下一小節展現。

各組件簡介

調度器

• 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格式兼容，最爲強大

使用

當你須要調度做業的時候，你須要爲這個做業選擇一個觸發器，用來描述該做業將在什麼時候被觸發，APScheduler有3中內置的觸發器類型：

• 新建一個調度器（scheduler）
• 添加一個調度任務（job store)
• 運行調度任務

添加做業

有兩種方式能夠添加一個新的做業：

• add_job來添加做業
• 裝飾器模式添加做業

只執行一次

import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
def job2(text):
    print('job2', datetime.datetime.now(), text)
scheduler = BlockingScheduler()
scheduler.add_job(job2, 'date', run_date=datetime.datetime(2019, 2, 25, 19, 5, 6), args=['text'], id='job2')
scheduler.start()

上例中，只在2010-2-25 19:05:06執行一次，args傳遞一個text參數。

間隔執行

下面來個簡單的例子，做業每一個5秒執行一次：

import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

def job1():
    print('job1', datetime.datetime.now())
scheduler = BlockingScheduler()
scheduler.add_job(job1, 'interval', seconds=5, id='job1')  # 每隔5秒執行一次
scheduler.start()

天天凌晨1點30分50秒執行一次

from apscheduler.schedulers.blocking import BlockingScheduler  # 後臺運行
sc = BlockingScheduler()
f = open('t1.text', 'a', encoding='utf8')
@sc.scheduled_job('cron', day_of_week='*', hour=1, minute='30', second='50')
def check_db():
    print(111111111111)
if __name__ == '__main__':
    try:
        sc.start()
        f.write('定時任務成功執行')
    except Exception as e:
        sc.shutdown()
        f.write('定時任務執行失敗')
    finally:
        f.close()

---

see also： APScheduler（Python化的Cron）使用總結 定時任務 | APScheduler