來(lái)源：https://www.biaodianfu.com/python-schedule.html
在日常工作中，我們常常會(huì )用到需要周期性執行的任務(wù)，一種方式是采用 Linux 系統自帶的 crond 結合命令行實(shí)現。另外一種方式是直接使用Python。接下來(lái)整理的是常見(jiàn)的Python定時(shí)任務(wù)的實(shí)現方式。
位于 time 模塊中的 sleep(secs) 函數，可以實(shí)現令當前執行的線(xiàn)程暫停 secs 秒后再繼續執行。所謂暫停，即令當前線(xiàn)程進(jìn)入阻塞狀態(tài)，當達到 sleep() 函數規定的時(shí)間后，再由阻塞狀態(tài)轉為就緒狀態(tài)，等待 CPU 調度。
基于這樣的特性我們可以通過(guò)while死循環(huán)+sleep()的方式實(shí)現簡(jiǎn)單的定時(shí)任務(wù)。
代碼示例：
import datetime
import time
def time_printer():
now = datetime.datetime.now()
ts = now.strftime('%Y-%m-%d %H:%M:%S')
print('do func time :', ts)
def loop_monitor():
while True:
time_printer()
time.sleep(5) # 暫停5秒
if __name__ == '__main__':
loop_monitor()
主要缺點(diǎn)：
Timeloop是一個(gè)庫，可用于運行多周期任務(wù)。這是一個(gè)簡(jiǎn)單的庫，它使用decorator模式在線(xiàn)程中運行標記函數。
示例代碼：
import time
from timeloop import Timeloop
from datetime import timedelta
tl = Timeloop()
@tl.job(interval=timedelta(seconds=2))
def sample_job_every_2s():
print '2s job current time : {}'.format(time.ctime())
@tl.job(interval=timedelta(seconds=5))
def sample_job_every_5s():
print '5s job current time : {}'.format(time.ctime())
@tl.job(interval=timedelta(seconds=10))
def sample_job_every_10s():
print '10s job current time : {}'.format(time.ctime())
threading 模塊中的 Timer 是一個(gè)非阻塞函數，比 sleep 稍好一點(diǎn)，timer最基本理解就是定時(shí)器，我們可以啟動(dòng)多個(gè)定時(shí)任務(wù)，這些定時(shí)器任務(wù)是異步執行，所以不存在等待順序執行問(wèn)題。
Timer(interval, function, args=[ ], kwargs={ })
代碼示例：
備注：Timer只能執行一次，這里需要循環(huán)調用，否則只能執行一次
sched模塊實(shí)現了一個(gè)通用事件調度器，在調度器類(lèi)使用一個(gè)延遲函數等待特定的時(shí)間，執行任務(wù)。同時(shí)支持多線(xiàn)程應用程序，在每個(gè)任務(wù)執行后會(huì )立刻調用延時(shí)函數，以確保其他線(xiàn)程也能執行。
class sched.scheduler(timefunc, delayfunc)這個(gè)類(lèi)定義了調度事件的通用接口，它需要外部傳入兩個(gè)參數，timefunc是一個(gè)沒(méi)有參數的返回時(shí)間類(lèi)型數字的函數(常用使用的如time模塊里面的time)，delayfunc應該是一個(gè)需要一個(gè)參數來(lái)調用、與timefunc的輸出兼容、并且作用為延遲多個(gè)時(shí)間單位的函數(常用的如time模塊的sleep)。
代碼示例：
import datetime
import time
import sched
def time_printer():
now = datetime.datetime.now()
ts = now.strftime('%Y-%m-%d %H:%M:%S')
print('do func time :', ts)
loop_monitor()
def loop_monitor():
s = sched.scheduler(time.time, time.sleep) # 生成調度器
s.enter(5, 1, time_printer, ())
s.run()
if __name__ == '__main__':
loop_monitor()
scheduler對象主要方法:
個(gè)人點(diǎn)評：比threading.Timer更好，不需要循環(huán)調用。
schedule是一個(gè)第三方輕量級的任務(wù)調度模塊，可以按照秒，分，小時(shí)，日期或者自定義事件執行時(shí)間。schedule允許用戶(hù)使用簡(jiǎn)單、人性化的語(yǔ)法以預定的時(shí)間間隔定期運行Python函數(或其它可調用函數)。
先來(lái)看代碼，是不是不看文檔就能明白什么意思？
import schedule
import time
def job():
print('I'm working...')
schedule.every(10).seconds.do(job)
schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at('10:30').do(job)
schedule.every(5).to(10).minutes.do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at('13:15').do(job)
schedule.every().minute.at(':17').do(job)
while True:
schedule.run_pending()
time.sleep(1)
裝飾器：通過(guò) @repeat() 裝飾靜態(tài)方法
import time
from schedule import every, repeat, run_pending
@repeat(every().second)
def job():
print('working...')
while True:
run_pending()
time.sleep(1)
傳遞參數：
import schedule
def greet(name):
print('Hello', name)
schedule.every(2).seconds.do(greet, name='Alice')
schedule.every(4).seconds.do(greet, name='Bob')
while True:
schedule.run_pending()
裝飾器同樣能傳遞參數：
from schedule import every, repeat, run_pending
@repeat(every().second, 'World')
@repeat(every().minute, 'Mars')
def hello(planet):
print('Hello', planet)
while True:
run_pending()
取消任務(wù)：
import schedule
i = 0
def some_task():
global i
i += 1
print(i)
if i == 10:
schedule.cancel_job(job)
print('cancel job')
exit(0)
job = schedule.every().second.do(some_task)
while True:
schedule.run_pending()
運行一次任務(wù)：
import time
import schedule
def job_that_executes_once():
print('Hello')
return schedule.CancelJob
schedule.every().minute.at(':34').do(job_that_executes_once)
while True:
schedule.run_pending()
time.sleep(1)
根據標簽檢索任務(wù)：
# 檢索所有任務(wù)：schedule.get_jobs()
import schedule
def greet(name):
print('Hello {}'.format(name))
schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')
schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')
schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')
schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')
friends = schedule.get_jobs('friend')
print(friends)
根據標簽取消任務(wù)：
# 取消所有任務(wù)：schedule.clear()
import schedule
def greet(name):
print('Hello {}'.format(name))
if name == 'Cancel':
schedule.clear('second-tasks')
print('cancel second-tasks')
schedule.every().second.do(greet, 'Andrea').tag('second-tasks', 'friend')
schedule.every().second.do(greet, 'John').tag('second-tasks', 'friend')
schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')
schedule.every(5).seconds.do(greet, 'Cancel').tag('daily-tasks', 'guest')
while True:
schedule.run_pending()
運行任務(wù)到某時(shí)間：
import schedule
from datetime import datetime, timedelta, time
def job():
print('working...')
schedule.every().second.until('23:59').do(job) # 今天23:59停止
schedule.every().second.until('2030-01-01 18:30').do(job) # 2030-01-01 18:30停止
schedule.every().second.until(timedelta(hours=8)).do(job) # 8小時(shí)后停止
schedule.every().second.until(time(23, 59, 59)).do(job) # 今天23:59:59停止
schedule.every().second.until(datetime(2030, 1, 1, 18, 30, 0)).do(job) # 2030-01-01 18:30停止
while True:
schedule.run_pending()
馬上運行所有任務(wù)（主要用于測試）：
import schedule
def job():
print('working...')
def job1():
print('Hello...')
schedule.every().monday.at('12:40').do(job)
schedule.every().tuesday.at('16:40').do(job1)
schedule.run_all()
schedule.run_all(delay_seconds=3) # 任務(wù)間延遲3秒
并行運行：使用 Python 內置隊列實(shí)現：
import threading
import time
import schedule
def job1():
print('I'm running on thread %s' % threading.current_thread())
def job2():
print('I'm running on thread %s' % threading.current_thread())
def job3():
print('I'm running on thread %s' % threading.current_thread())
def run_threaded(job_func):
job_thread = threading.Thread(target=job_func)
job_thread.start()
schedule.every(10).seconds.do(run_threaded, job1)
schedule.every(10).seconds.do(run_threaded, job2)
schedule.every(10).seconds.do(run_threaded, job3)
while True:
schedule.run_pending()
time.sleep(1)
APScheduler（advanceded python scheduler）基于Quartz的一個(gè)Python定時(shí)任務(wù)框架，實(shí)現了Quartz的所有功能，使用起來(lái)十分方便。提供了基于日期、固定時(shí)間間隔以及crontab類(lèi)型的任務(wù)，并且可以持久化任務(wù)?；谶@些功能，我們可以很方便的實(shí)現一個(gè)Python定時(shí)任務(wù)系統。
它有以下三個(gè)特點(diǎn)：
APScheduler有四種組成部分：
示例代碼：
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
# 輸出時(shí)間
def job():
print(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
# BlockingScheduler
sched = BlockingScheduler()
sched.add_job(my_job, 'interval', seconds=5, id='my_job_id')
sched.start()
Job作為APScheduler最小執行單位。創(chuàng )建Job時(shí)指定執行的函數，函數中所需參數，Job執行時(shí)的一些設置信息。
構建說(shuō)明：
Trigger綁定到Job，在scheduler調度篩選Job時(shí)，根據觸發(fā)器的規則計算出Job的觸發(fā)時(shí)間，然后與當前時(shí)間比較確定此Job是否會(huì )被執行，總之就是根據trigger規則計算出下一個(gè)執行時(shí)間。
目前APScheduler支持觸發(fā)器：
觸發(fā)器參數：date
date定時(shí)，作業(yè)只執行一次。
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])
sched.add_job(my_job, 'date', run_date=datetime(2019, 7, 6, 16, 30, 5), args=['text'])
觸發(fā)器參數：interval
interval間隔調度
sched.add_job(job_function, 'interval', hours=2)
觸發(fā)器參數：cron
cron調度
CronTrigger可用的表達式：
| 表達式 | 參數類(lèi)型 | 描述 |
|---|---|---|
| * | 所有 | 通配符。例：minutes=*即每分鐘觸發(fā) |
| * / a | 所有 | 每隔時(shí)長(cháng)a執行一次。例：minutes=”* / 3″ 即每隔3分鐘執行一次 |
| a – b | 所有 | a – b的范圍內觸發(fā)。例：minutes=“2-5”。即2到5分鐘內每分鐘執行一次 |
| a – b / c | 所有 | a – b范圍內，每隔時(shí)長(cháng)c執行一次。 |
| xth y | 日 | 第幾個(gè)星期幾觸發(fā)。x為第幾個(gè)，y為星期幾 |
| last x | 日 | 一個(gè)月中，最后一個(gè)星期的星期幾觸發(fā) |
| last | 日 | 一個(gè)月中的最后一天觸發(fā) |
| x, y, z | 所有 | 組合表達式，可以組合確定值或上述表達式 |
# 6-8,11-12月第三個(gè)周五 00:00, 01:00, 02:00, 03:00運行
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
# 每周一到周五運行 直到2024-05-30 00:00:00
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2024-05-30'
Executor在scheduler中初始化，另外也可通過(guò)scheduler的add_executor動(dòng)態(tài)添加Executor。每個(gè)executor都會(huì )綁定一個(gè)alias，這個(gè)作為唯一標識綁定到Job，在實(shí)際執行時(shí)會(huì )根據Job綁定的executor找到實(shí)際的執行器對象，然后根據執行器對象執行Job。
Executor的種類(lèi)會(huì )根據不同的調度來(lái)選擇，如果選擇AsyncIO作為調度的庫，那么選擇AsyncIOExecutor，如果選擇tornado作為調度的庫，選擇TornadoExecutor，如果選擇啟動(dòng)進(jìn)程作為調度，選擇ThreadPoolExecutor或者ProcessPoolExecutor都可以。
Executor的選擇需要根據實(shí)際的scheduler來(lái)選擇不同的執行器。目前APScheduler支持的Executor：
Jobstore在scheduler中初始化，另外也可通過(guò)scheduler的add_jobstore動(dòng)態(tài)添加Jobstore。每個(gè)jobstore都會(huì )綁定一個(gè)alias，scheduler在A(yíng)dd Job時(shí)，根據指定的jobstore在scheduler中找到相應的jobstore，并將job添加到j(luò )obstore中。作業(yè)存儲器決定任務(wù)的保存方式， 默認存儲在內存中（MemoryJobStore），重啟后就沒(méi)有了。APScheduler支持的任務(wù)存儲器有：
不同的任務(wù)存儲器可以在調度器的配置中進(jìn)行配置（見(jiàn)調度器）
Event是APScheduler在進(jìn)行某些操作時(shí)觸發(fā)相應的事件，用戶(hù)可以自定義一些函數來(lái)監聽(tīng)這些事件，當觸發(fā)某些Event時(shí)，做一些具體的操作。常見(jiàn)的比如。Job執行異常事件 EVENT_JOB_ERROR。Job執行時(shí)間錯過(guò)事件 EVENT_JOB_MISSED。
目前APScheduler定義的Event：
Listener表示用戶(hù)自定義監聽(tīng)的一些Event，比如當Job觸發(fā)了EVENT_JOB_MISSED事件時(shí)可以根據需求做一些其他處理。
Scheduler是APScheduler的核心，所有相關(guān)組件通過(guò)其定義。scheduler啟動(dòng)之后，將開(kāi)始按照配置的任務(wù)進(jìn)行調度。除了依據所有定義Job的trigger生成的將要調度時(shí)間喚醒調度之外。當發(fā)生Job信息變更時(shí)也會(huì )觸發(fā)調度。
APScheduler支持的調度器方式如下，比較常用的為BlockingScheduler和BackgroundScheduler
Scheduler添加job流程：
Scheduler調度流程：
Celery是一個(gè)簡(jiǎn)單，靈活，可靠的分布式系統，用于處理大量消息，同時(shí)為操作提供維護此類(lèi)系統所需的工具, 也可用于任務(wù)調度。Celery 的配置比較麻煩，如果你只是需要一個(gè)輕量級的調度工具，Celery 不會(huì )是一個(gè)好選擇。
Celery 是一個(gè)強大的分布式任務(wù)隊列，它可以讓任務(wù)的執行完全脫離主程序，甚至可以被分配到其他主機上運行。我們通常使用它來(lái)實(shí)現異步任務(wù)（async task）和定時(shí)任務(wù)（crontab）。異步任務(wù)比如是發(fā)送郵件、或者文件上傳, 圖像處理等等一些比較耗時(shí)的操作 ，定時(shí)任務(wù)是需要在特定時(shí)間執行的任務(wù)。
需要注意，celery本身并不具備任務(wù)的存儲功能，在調度任務(wù)的時(shí)候肯定是要把任務(wù)存起來(lái)的，因此在使用celery的時(shí)候還需要搭配一些具備存儲、訪(fǎng)問(wèn)功能的工具，比如：消息隊列、Redis緩存、數據庫等。官方推薦的是消息隊列RabbitMQ，有些時(shí)候使用Redis也是不錯的選擇。
它的架構組成如下圖：
Celery架構，它采用典型的生產(chǎn)者-消費者模式，主要由以下部分組成：
實(shí)際應用中，用戶(hù)從Web前端發(fā)起一個(gè)請求，我們只需要將請求所要處理的任務(wù)丟入任務(wù)隊列broker中，由空閑的worker去處理任務(wù)即可，處理的結果會(huì )暫存在后臺數據庫backend中。我們可以在一臺機器或多臺機器上同時(shí)起多個(gè)worker進(jìn)程來(lái)實(shí)現分布式地并行處理任務(wù)。
Celery定時(shí)任務(wù)實(shí)例：
Apache Airflow 是Airbnb開(kāi)源的一款數據流程工具，目前是Apache孵化項目。以非常靈活的方式來(lái)支持數據的ETL過(guò)程，同時(shí)還支持非常多的插件來(lái)完成諸如HDFS監控、郵件通知等功能。Airflow支持單機和分布式兩種模式，支持Master-Slave模式，支持Mesos等資源調度，有非常好的擴展性。被大量公司采用。
Airflow使用Python開(kāi)發(fā)，它通過(guò)DAGs(Directed Acyclic Graph, 有向無(wú)環(huán)圖)來(lái)表達一個(gè)工作流中所要執行的任務(wù)，以及任務(wù)之間的關(guān)系和依賴(lài)。比如，如下的工作流中，任務(wù)T1執行完成，T2和T3才能開(kāi)始執行，T2和T3都執行完成，T4才能開(kāi)始執行。
Airflow提供了各種Operator實(shí)現，可以完成各種任務(wù)實(shí)現：
除了以上這些 Operators 還可以方便的自定義 Operators 滿(mǎn)足個(gè)性化的任務(wù)需求。
一些情況下，我們需要根據執行結果執行不同的任務(wù)，這樣工作流會(huì )產(chǎn)生分支。如:
這種需求可以使用BranchPythonOperator來(lái)實(shí)現。
通常，在一個(gè)運維系統，數據分析系統，或測試系統等大型系統中，我們會(huì )有各種各樣的依賴(lài)需求。包括但不限于：
crontab 可以很好地處理定時(shí)執行任務(wù)的需求，但僅能管理時(shí)間上的依賴(lài)。Airflow 的核心概念 DAG（有向無(wú)環(huán)圖）—— 來(lái)表現工作流。
在一個(gè)可擴展的生產(chǎn)環(huán)境中，Airflow 含有以下組件：
Worker的具體實(shí)現由配置文件中的executor來(lái)指定，airflow支持多種Executor:
生產(chǎn)環(huán)境一般使用CeleryExecutor和KubernetesExecutor。
使用CeleryExecutor的架構如圖:
使用KubernetesExecutor的架構如圖:
其它參考：
聯(lián)系客服
