Python 實現定時任務（sleep、Timer 、sched、APScheduler）

一、循環sleep

from datetime import datetime
import time
# 每n秒執行一次
def timer(n):
    while True:
        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
        time.sleep(n)
# 2s
timer(2)

缺點：sleep是一個阻塞函數，只能執行固定間隔時間的任務，無法完成定時任務（在sleep的這一段時間，啥都不能做）

 

二、threading模塊中的Timer

from datetime import datetime
from threading import Timer
# 打印時間函數
def printTime(inc):
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    t = Timer(inc, printTime, (inc,))
    t.start()
# 2s
printTime(2)

 

Timer函數：第一個參數是時間間隔（單位是秒），第二個參數是要調用的函數名，第三個參數是調用函數的參數(tuple)

缺點：threading 模塊中的 Timer 是一個非阻塞函數，無法完成定時任務

 

三、使用sched內置模塊

import sched
import time
from datetime import datetime
# 初始化sched模塊的 scheduler 類
# 第一個參數是一個可以返回時間戳的函數，第二個參數可以在定時未到達之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)
# 被周期性調度觸發的函數
def printTime(inc):
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    schedule.enter(inc, 0, printTime, (inc,))
# 默認參數60s
def main(inc=60):
    # enter四個參數分別為：間隔事件、優先級（用於同時間到達的兩個事件同時執行時定序）、被調用觸發的函數，
    # 給該觸發函數的參數（tuple形式）
    schedule.enter(0, 0, printTime, (inc,))
    schedule.run()
# 10s 輸出一次
main(10)

sched 模塊是 Python 內置的模塊，它是一個調度（延時處理機制），每次想要定時執行某任務都必須寫入一個調度。

sched 使用步驟如下：

1）生成調度器：
s = sched.scheduler(time.time,time.sleep)
第一個參數是一個可以返回時間戳的函數，第二個參數可以在定時未到達之前阻塞。

（2）加入調度事件
其實有 enter、enterabs 等等，我們以 enter 為例子。
s.enter(x1,x2,x3,x4)
四個參數分別為：間隔事件、優先級（用於同時間到達的兩個事件同時執行時定序）、被調用觸發的函數，給觸發函數的參數（注意：一定要以 tuple 給，如果只有一個參數就(xx,)）

（3）運行
s.run()
注意 sched 模塊不是循環的，一次調度被執行后就 Over 了，如果想再執行，請再次 enter

 四、APScheduler定時框架(原文鏈接)

APScheduler是一個 Python 定時任務框架，提供了基於日期、固定時間間隔以及 crontab 類型的任務，並且可以持久化任務、並以 daemon 方式運行應用。

安裝：

$ pip install apscheduler

 案例：周一到周五每天早上6點半執行任務

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
# 輸出時間
def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# BlockingScheduler
scheduler = BlockingScheduler()
scheduler.add_job(job, 'cron', day_of_week='1-5', hour=6, minute=30)
scheduler.start()

BlockingScheduler 是 APScheduler 中的調度器，APScheduler 中有兩種常用的調度器，BlockingScheduler 和 BackgroundScheduler，當調度器是應用中唯一要運行的任務時，使用 BlockingSchedule，如果希望調度器在后台執行，使用 BackgroundScheduler。

BlockingSchedule：當調度器是應用中唯一要運行的任務時

BackgroundScheduler：如果希望調度器在后台執行，使用 BackgroundScheduler

APScheduler 四個組件分別為：觸發器(trigger)，作業存儲(job store)，執行器(executor)，調度器(scheduler)。

觸發器(trigger)：包含調度邏輯，每一個作業有它自己的觸發器，用於決定接下來哪一個作業會運行。

APScheduler 有三種內建的 trigger:
date: 特定的時間點觸發：

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

• run_date: 在某天執行任務
• timezone: 在某段時間執行任務

from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
def my_job(text):
    print(text)
# The job will be executed on November 6th, 2009
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])
sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text'])
sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05', args=['text'])
# The 'date' trigger and datetime.now() as run_date are implicit
sched.add_job(my_job, args=['text'])
sched.start()

interval: 固定時間間隔觸發：

• weeks: 每隔幾周執行一次 |　weeks=0
• days: 每隔幾天執行一次 | days=0
• hours: 每隔幾小時執行一次 | hours=0
• minutes: 每隔幾分執行一次 | minutes=0
• seconds: 每隔幾秒執行一次 | seconds=0
• start_date: 最早執行時間 | start_date=None
• end_date: 最晚執行時間 | end_date=None
• timezone: 執行時間區間 | timezone=None

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

def job_function():
    print("Hello World")
# BlockingScheduler
sched = BlockingScheduler()
# Schedule job_function to be called every two hours
sched.add_job(job_function, 'interval', hours=2)
# The same as before, but starts on 2010-10-10 at 9:30 and stops on 2014-06-15 at 11:00
sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00')
sched.start()

cron: 在特定時間周期性地觸發：

• year: 4位數字
• month: 月 (1-12)
• day: 天 (1-31)
• week: 標准周 (1-53)
• day_of_week: 周中某天 (0-6 or mon,tue,wed,thu,fri,sat,sun)
• hour: 小時 (0-23)
• minute:分鍾 (0-59)
• second: 秒 (0-59)
• start_date: 最早執行時間
• end_date: 最晚執行時間
• timezone: 執行時間區間

表達式:

 

from apscheduler.schedulers.blocking import BlockingScheduler

def job_function():
    print("Hello World")
# BlockingScheduler
sched = BlockingScheduler()
# Schedules job_function to be run on the third Friday
# of June, July, August, November and December at 00:00, 01:00, 02:00 and 03:00
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
# Runs from Monday to Friday at 5:30 (am) until 2014-05-30 00:00:00
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')
sched.start()

 

作業存儲(job store)：
存儲被調度的作業，默認的作業存儲是簡單地把作業保存在內存中，其他的作業存儲是將作業保存在數據庫中。一個作業的數據講在保存在持久化作業存儲時被序列化，並在加載時被反序列化。調度器不能分享同一個作業存儲。
APScheduler 默認使用 MemoryJobStore，可以修改使用 DB 存儲方案

執行器(executor)：

處理作業的運行，他們通常通過在作業中提交制定的可調用對象到一個線程或者進程池來進行。當作業完成時，執行器將會通知調度器。
最常用的 executor 有兩種：
ProcessPoolExecutor（進程池）
ThreadPoolExecutor（線程池，max:10）

調度器(scheduler)：
通常在應用中只有一個調度器，應用的開發者通常不會直接處理作業存儲、調度器和觸發器，相反，調度器提供了處理這些的合適的接口。配置作業存儲和執行器可以在調度器中完成，例如添加、修改和移除作業。

配置調度器：
APScheduler提供了許多不同的方式來配置調度器，你可以使用一個配置字典或者作為參數關鍵字的方式傳入。

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 定義BlockingScheduler
sched = BlockingScheduler()
sched.add_job(job, 'interval', seconds=5)
sched.start()

# 設置 job store(使用mongo存儲)和 executor

from datetime import datetime
from pymongo import MongoClient
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
# MongoDB 參數
host = '127.0.0.1'
port = 27017
client = MongoClient(host, port)
# 輸出時間
def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 存儲方式
jobstores = {
    'mongo': MongoDBJobStore(collection='job', database='test', client=client),
    'default': MemoryJobStore()
}
executors = {
    'default': ThreadPoolExecutor(10),
    'processpool': ProcessPoolExecutor(3)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
scheduler.add_job(job, 'interval', seconds=5, jobstore='mongo')
scheduler.start()

添加job：
add_job()
scheduled_job()
第一種方法返回一個apscheduler.job.Job 的實例，可以用來改變或者移除 job，第二種方法只適用於應用運行期間不會改變的 job。

from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
# 裝飾器
@sched.scheduled_job('interval', id='my_job_id', seconds=5)
def job_function():
    print("Hello World")
# 開始
sched.start()

 

移除 job：
remove_job 使用 jobID 移除
job.remove() 使用 add_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')

暫停一個 job：

apscheduler.job.Job.pause()
apscheduler.schedulers.base.BaseScheduler.pause_job()

恢復一個 job：

apscheduler.job.Job.resume()
apscheduler.schedulers.base.BaseScheduler.resume_job()

獲取 job 列表：

apscheduler.get_jobs()

修改job：

 apscheduler.job.Job.modify() 或者 modify_job() 修改一個 job 的屬性

job.modify(max_instances=6, name='Alternate name')
modify_job('my_job_id', trigger='cron', minute='*/5')

關閉job:

默認情況下調度器會等待所有的 job 完成后，關閉所有的調度器和作業存儲。將 wait 選項設置為 False 可以立即關閉。

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

scheduler 可以添加事件監聽器，並在特殊的時間觸發

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)