摘要：

今天構(gòu)建的大多數(shù)應(yīng)用程序都需要某種方式的調(diào)度機(jī)制。輪詢(xún) API 或數(shù)據(jù)庫(kù)、不斷檢查系統(tǒng)健康狀況、將日志存檔等是常見(jiàn)的例子。 Kubernetes和Apache Mesos等使用自動(dòng)伸縮擴(kuò)容技術(shù)（Auto-scaling）的軟件需要檢查部署的應(yīng)用程序的狀態(tài)，為此它們使用定期運(yùn)行的存活探針（Liveness Probe）。調(diào)度任務(wù)需要與業(yè)務(wù)邏輯解耦，因此我們要使用解耦的執(zhí)行隊(duì)列，例如Redis隊(duì)列。

Python 有幾種方法可以定時(shí)調(diào)度一個(gè)任務(wù)，這就是我們將在本文中學(xué)習(xí)的內(nèi)容。我將使用以下方式討論調(diào)度任務(wù)：

• 簡(jiǎn)單循環(huán) （Simple Loops）
• 簡(jiǎn)單循環(huán)但是使用了線程 （Simple Loops but Threaded）
• 調(diào)度庫(kù) （Schedule Library）
• Python Crontab

• RQ 調(diào)度器作為解耦隊(duì)列 （RQ Scheduler as decoupled queues）
  

1、簡(jiǎn)單循環(huán) Simple loops

使用簡(jiǎn)單循環(huán)來(lái)實(shí)現(xiàn)調(diào)度任務(wù)這是毫不費(fèi)力的。使用無(wú)限運(yùn)行的 while 循環(huán)定期調(diào)用函數(shù)可用于調(diào)度作業(yè)，但這不是最好的方法，不過(guò)它是很有效的。可以使用內(nèi)置time模塊的slleep()來(lái)延遲執(zhí)行。不過(guò)這并不是大多數(shù)作業(yè)的調(diào)度方式，因?yàn)?，它看起?lái)很難看，而且與其他方法相比，它的可讀性較差。

import time
​
def task():
 print("Job Completed!")
​
 while 1:
 task()
 time.sleep(10)

當(dāng)涉及到每天早上 9:00 或每周三晚上 7:45 等這些日程安排時(shí)，事情就變得比較棘手了。

import datetime
​
def task():
 print("Job Completed!")
​
 while 1:
 now = datetime.datetime.now()
 # schedule at every wednesday,7:45 pm
 if now.weekday == 3 and now.strftime("%H:%m") == "19:45":
  task()
 # sleep for 6 days
 time.sleep(6 * 24 * 60 * 60)

這是我的第一時(shí)間想到的解決辦法，不用謝！這種方法的一個(gè)問(wèn)題是這里的邏輯是阻塞的，即一旦在 python 項(xiàng)目中發(fā)現(xiàn)這段代碼，它就會(huì)卡在 while 1 循環(huán)中，從而阻塞其他代碼的執(zhí)行。

2、簡(jiǎn)單循環(huán)但是使用了線程Simple loops but threaded

線程是計(jì)算機(jī)科學(xué)中的一個(gè)概念。具有自己指令的小程序由進(jìn)程執(zhí)行并獨(dú)立管理，這就可以解決我們第一種方法的阻塞情況，讓我們看看怎么樣。

import time
import threading
​
def task():
 print("Job Completed!")
​
def schedule():
 while 1:
  task()
  time.sleep(10)
​
# makes our logic non blocking
thread = threading.Thread(target=schedule)
thread.start()

線程啟動(dòng)后，其底層邏輯無(wú)法被主線程修改，因此我們可能需要添加資源，程序通過(guò)這些資源可以檢查特定場(chǎng)景并根據(jù)它們執(zhí)行邏輯。

3、定時(shí)調(diào)度庫(kù) Schedule Library

早些時(shí)候，我說(shuō)使用 while 循環(huán)進(jìn)行調(diào)度看起來(lái)很丑陋，調(diào)度庫(kù)可以解決這個(gè)問(wèn)題。

import schedule
import time
​
def task():
 print("Job Executing!")
​
# for every n minutes
schedule.every(10).minutes.do(task)
​
# every hour
schedule.every().hour.do(task)
​
# every daya at specific time
schedule.every().day.at("10:30").do(task)
​
# schedule by name of day
schedule.every().monday.do(task)
​
# name of day with time
schedule.every().wednesday.at("13:15").do(task)
​
while True:
 schedule.run_pending()
 time.sleep(1)

正如您所見(jiàn)，通過(guò)這樣我們可以毫不費(fèi)力地創(chuàng)建多個(gè)調(diào)度計(jì)劃。我特別喜歡創(chuàng)建作業(yè)的方式和方法鏈（Method Chaining），另一方面，這個(gè)片段有一個(gè) while 循環(huán)，這意味著代碼被阻塞，不過(guò)我相信你已經(jīng)知道什么可以幫助我們解決這個(gè)問(wèn)題。

4、Python Crontab

Liunx 中的 crontab 實(shí)用程序是一種易于使用且被廣泛接受的調(diào)度解決方案。Python 庫(kù)python-crontab提供了一個(gè) API 來(lái)使用 Python 中的 CLI 工具。在crontab中，一個(gè)定時(shí)調(diào)度使用 unix-cron字符串格式（ *）來(lái)描述，它是一組五個(gè)值的一條線，這表明當(dāng)作業(yè)應(yīng)該被執(zhí)行時(shí)，python-crontab 將在文件中寫(xiě)入 crontab 的計(jì)劃轉(zhuǎn)換為寫(xiě)入編程方法。

from crontab import CronTab
​
cron = CronTab(user='root')
​
job = cron.new(command='my_script.sh')
​
job.hour.every(1)
cron.write()

python-crontab 不會(huì)自動(dòng)保存計(jì)劃，需要執(zhí)行 write() 方法來(lái)保存計(jì)劃。還有更多功能，我強(qiáng)烈建議您查看他們的文檔。

5、RQ 調(diào)度器 RQ Scheduler

有些任務(wù)不能立即執(zhí)行，因此我們需要根據(jù) LIFO 或 FIFO 等隊(duì)列系統(tǒng)創(chuàng)建任務(wù)隊(duì)列并彈出任務(wù)。python-rq允許我們做到這一點(diǎn)，使用 Redis 作為代理來(lái)排隊(duì)作業(yè)。新作業(yè)的條目存儲(chǔ)為帶有信息的哈希映射，例如created_at, enqueued_at, origin, data, description.

排隊(duì)任務(wù)由名為 worker 的程序執(zhí)行。workers 在 Redis 緩存中也有一個(gè)條目，負(fù)責(zé)將任務(wù)出列以及更新 Redis 中的任務(wù)狀態(tài)。任務(wù)可以在需要時(shí)排隊(duì)，但要安排它們，我們需要rq-scheduler。

from rq_scheduler import Scheduler
​
queue = Queue('circle', connection=Redis())
scheduler = Scheduler(queue=queue)
​
scheduler.schedule(
 scheduled_time=datetime.utcnow(), # Time for first execution, in UTC timezone
 func=func,# Function to be queued
 args=[arg1, arg2], # Arguments passed into function when executed
 kwargs={'foo': 'bar'},# Keyword arguments passed into function when executed
 interval=60, # Time before the function is called again, in seconds
 repeat=None,# Repeat this number of times (None means repeat forever)
 meta={'foo': 'bar'}# Arbitrary pickleable data on the job itself
)

RQ worker（RQ 工作器）必須在終端中單獨(dú)啟動(dòng)或通過(guò) python-rq 工作器啟動(dòng)。一旦任務(wù)被觸發(fā)，就可以在工作終端中看到，在成功和失敗場(chǎng)景中都可以使用單獨(dú)的函數(shù)回調(diào)。

6、總結(jié) Conclusion

還有一些用于調(diào)度的庫(kù)，但在這里，我已經(jīng)討論了最常見(jiàn)的庫(kù)。值得一提的是Celery，celery 的另一個(gè)優(yōu)點(diǎn)是用戶(hù)可以在多個(gè)代理之間進(jìn)行選擇。我很感激你讀到最后。也可以看看我的其他文章。干杯!

到此這篇關(guān)于Python使用定時(shí)調(diào)度任務(wù)的方式的文章就介紹到這了,更多相關(guān)Python使用定時(shí)調(diào)度任務(wù)方式內(nèi)容請(qǐng)搜索本站以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持本站！

版權(quán)聲明：本站文章來(lái)源標(biāo)注為YINGSOO的內(nèi)容版權(quán)均為本站所有，歡迎引用、轉(zhuǎn)載，請(qǐng)保持原文完整并注明來(lái)源及原文鏈接。禁止復(fù)制或仿造本網(wǎng)站，禁止在非www.sddonglingsh.com所屬的服務(wù)器上建立鏡像，否則將依法追究法律責(zé)任。本站部分內(nèi)容來(lái)源于網(wǎng)友推薦、互聯(lián)網(wǎng)收集整理而來(lái)，僅供學(xué)習(xí)參考，不代表本站立場(chǎng)，如有內(nèi)容涉嫌侵權(quán)，請(qǐng)聯(lián)系alex-e#qq.com處理。