Python定時任務神器-APScheduler

• 2019 年 10 月 4 日
• 筆記

定時任務在很多的開發場景中都會使用到，在Python中也提供很多的定時任務庫。比如：

• sched
• schedule
• celery

但是這些定時任務庫都只是提供了簡答的，或者只支援靜態的定時任務。而對於需要複雜定時功能，或者動態註冊定時任務的場景，則無法滿足。

而今天介紹的主角 – APScheduler，則會完美的解決這個問題。

• 多種的定時任務類型支援
• 靜態、動態定時任務支援

簡單說明

不管你使用哪種APScheduler的定時任務，你都需要先了解APScheduler的簡單機制。即：job、executors、jobstores、trigger、scheduler等

job

即需要被執行的具體任務，主要對應Python中的函數或方法。在APScheduler中即可提前配置，也可以動態添加job。

executors

即執行job的對象。通常可以是多執行緒、多進程、協程等對象。

jobstores

即存儲job元數據的地方。可以是memory、sqllite、mysql等。

trigger

即決定任務的觸發模式。通常有指定時間、指定時間間隔、指定周期策略等。

scheduler

用於調度和管理上述提到的所有對象。

任意一個APScheduler的實例啟動的時候都需要配置這些初始參數，如果沒有指定則會使用默認的值。

使用方式

首先你得安裝apscheduler，方式如下：

pip install apscheduler  

靜態配置任務

import time  from apscheduler.schedulers.blocking import BlockingScheduler    sched = BlockingScheduler()    @sched.scheduled_job('interval', seconds=5)     # 以5s為間隔來觸發  def my_job():      print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))    sched.start()  

動態添加任務

import time  from apscheduler.schedulers.blocking import BlockingScheduler    def my_job():      print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))    sched = BlockingScheduler()  # 在11,12月份的第三個星期五的00:00,01:00,02:00,03:00觸發  sched.add_job(my_job, 'cron', month='11-12', day='3rd fri', hour='0-3')    sched.start()  

非同步任務

上面的任務都是阻塞任務，即任務執行過程中需要持續等待結果；如果你不希望等待結果的話，則可以使用非阻塞的非同步任務。

import time  import datetime  from pytz import timezone  from apscheduler.schedulers.background import BackgroundScheduler  from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore  from apscheduler.executors.pool import ThreadPoolExecutor    def my_job(name):      print('%s: %s' % (time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())), name))    jobstores = {      'default': SQLAlchemyJobStore(url='sqlite:////tmp/test.db')  }  executors = {      'default': ThreadPoolExecutor(20)  }  job_defaults = {      'coalesce': False,      'max_instances': 3  }  task_scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults,                                       timezone=timezone('Asia/Shanghai'))  task_scheduler.start()  # 指定特定的日期觸發任務，這類任務只會觸發一次  job = task_scheduler.add_job(my_job, 'date', run_date=datetime(2019, 3, 6, 9, 47, 5), args=('date',))  

其它使用API

除了註冊和添加任務之外，apscheduler還提供了其它比較友好的API介面。比如：

• 暫停任務
• 啟動任務
• 刪除任務

job = scheduler.add_job(myfunc, 'interval', minutes=2)  job.pause()  job.resume()  job.remove()