Python中輕量級(jí)定時(shí)任務(wù)調(diào)度庫schedule使用指南
schedule是一款專為簡(jiǎn)化定時(shí)任務(wù)調(diào)度而設(shè)計(jì)的Python庫，它通過直觀的語法降低了周期性任務(wù)的實(shí)現(xiàn)門檻。作為進(jìn)程內(nèi)調(diào)度器，它無需額外守護(hù)進(jìn)程，輕量且無外部依賴，適合快速搭建自動(dòng)化任務(wù)。不過，該庫在功能完整性上有所取舍，目前暫不支持?jǐn)帱c(diǎn)續(xù)傳、亞秒級(jí)精度控制以及多任務(wù)并行執(zhí)行等復(fù)雜場(chǎng)景。
schedule庫的官方倉(cāng)庫地址見：schedule，schedule庫的官方文檔見：schedule-doc。
schedule庫支持在Python 3.7及以上版本的環(huán)境中運(yùn)行，schedule庫的安裝命令如下：
pip install schedule
1 使用入門
1.1 基礎(chǔ)使用
1.1.1 相對(duì)調(diào)用
schedule庫通過創(chuàng)建調(diào)度器，設(shè)置時(shí)間單位，注冊(cè)待調(diào)用函數(shù)，返回任務(wù)對(duì)象實(shí)現(xiàn)任務(wù)周期調(diào)用。例如：
# 每3秒執(zhí)行一次任務(wù) schedule.every(3).seconds.do(job)
這種調(diào)用方式是相對(duì)調(diào)用方式，任務(wù)執(zhí)行間隔是相對(duì)于當(dāng)前時(shí)間點(diǎn)計(jì)算的。例如17:59:01創(chuàng)建調(diào)度器，下一次執(zhí)行時(shí)間為17:59:04。
示例代碼如下：
import schedule
import time
from datetime import datetime
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
def job():
"""定時(shí)執(zhí)行的任務(wù)函數(shù)"""
print(f"{get_now_time()} I'm working...")
print(f"調(diào)度器創(chuàng)建時(shí)間：{get_now_time()}")
# 調(diào)度器返回job對(duì)象
schedule_job = schedule.every(3).seconds.do(job)
print(f"調(diào)度器下一次運(yùn)行時(shí)間：{schedule_job.next_run}")
# 任務(wù)調(diào)度主循環(huán)
# 持續(xù)檢查是否有待執(zhí)行的任務(wù)
while True:
schedule.run_pending() # 檢查并執(zhí)行待執(zhí)行的任務(wù)
time.sleep(1) # 休眠1秒避免CPU占用過高
# 為None就是沒運(yùn)行
print(f"調(diào)度器上一次運(yùn)行時(shí)間：{schedule_job.last_run}")
print(f"調(diào)度器下一次運(yùn)行時(shí)間：{schedule_job.next_run}")
其他相對(duì)時(shí)間間隔調(diào)用代碼如下：
# 每3分鐘執(zhí)行一次任務(wù) schedule.every(3).minutes.do(job) # 每小時(shí)執(zhí)行一次任務(wù) schedule.every().hours.do(job) # 每3天執(zhí)行一次任務(wù) schedule.every(3).days.do(job) # 每3周執(zhí)行一次任務(wù) schedule.every(3).weeks.do(job) # 每周一執(zhí)行任務(wù) schedule.every().monday.do(job)
1.1.2 絕對(duì)調(diào)用
schedule可以在相對(duì)調(diào)用的基礎(chǔ)上結(jié)合at函數(shù)實(shí)現(xiàn)絕對(duì)時(shí)間的調(diào)度。例如：
schedule.every(3).hours.at("11:16").do(job)
上述代碼可拆解為：
job1 = schedule.every(3).hours
job2 = job1.at("11:16").do(job)
此處的job1基于當(dāng)前調(diào)用器創(chuàng)建時(shí)間（例如18:34:54），以小時(shí)為間隔粒度進(jìn)行設(shè)定，即每間隔3小時(shí)執(zhí)行一次任務(wù)，因此下一次任務(wù)執(zhí)行時(shí)間為21:34:54。而at()函數(shù)的作用是在job1設(shè)定的時(shí)間粒度和間隔范圍內(nèi)，具體指定分鐘和秒。以job2中的at("11:16")為例，這里的11:16代表分鐘和秒。它會(huì)在18:34:54至21:34:54的時(shí)間區(qū)間內(nèi)，定位離21:34:54最近的11分16秒時(shí)刻，即21:11:16。
若按每分鐘調(diào)用一次任務(wù)，可通過at指定固定執(zhí)行秒數(shù)。例如，當(dāng)創(chuàng)建時(shí)間為18:34:54時(shí)，下一次執(zhí)行時(shí)間為18:35:16，對(duì)應(yīng)代碼如下：
schedule_job = schedule.every().minutes.at(":16").do(job)
at函數(shù)的輸入范圍由設(shè)定的時(shí)間粒度決定，且僅支持日級(jí)、時(shí)級(jí)、分級(jí)三類時(shí)間粒度，具體子粒度支持規(guī)則如下：
- 每日任務(wù)：支持HH:MM:SS（時(shí)分秒）和HH:MM兩種格式（HH:MM默認(rèn)補(bǔ)全為HH:MM:00）；
- 每小時(shí)任務(wù)：支持MM:SS（分秒）和:MM兩種格式（:MM默認(rèn)補(bǔ)全為MM:00）；
- 每分鐘任務(wù)：僅支持:SS（秒）格式。
其他絕對(duì)時(shí)間間隔調(diào)用代碼如下：
# 基于當(dāng)前時(shí)間，每分鐘的16秒執(zhí)行任務(wù)
schedule.every().minutes.at(":16").do(job)
# 基于當(dāng)前時(shí)間，每小時(shí)的第23分鐘00秒執(zhí)行任務(wù)
schedule.every().hours.at(":23").do(job)
# 基于當(dāng)前時(shí)間，每5小時(shí)的第20分30秒執(zhí)行任務(wù)
schedule.every(5).hours.at("20:30").do(job)
# 基于當(dāng)前時(shí)間，每天上午10:30:00執(zhí)行任務(wù)
schedule.every().days.at("10:30").do(job)
# 基于當(dāng)前時(shí)間，每天上午10:30:42執(zhí)行任務(wù)
schedule.every().days.at("10:30:42").do(job)
# 基于當(dāng)前時(shí)間，每天上午12:42（阿姆斯特丹時(shí)區(qū)）執(zhí)行任務(wù)
schedule.every().days.at("12:42", "Europe/Amsterdam").do(job)
# 基于當(dāng)前時(shí)間，每周三下午1點(diǎn)15分執(zhí)行任務(wù)
schedule.every().wednesday.at("13:15").do(job)
以下示例代碼演示了多任務(wù)調(diào)度場(chǎng)景。作為輕量級(jí)任務(wù)調(diào)度庫，schedule會(huì)維護(hù)任務(wù)列表，調(diào)用schedule.run_pending()時(shí)檢查所有任務(wù)觸發(fā)條件，滿足條件的任務(wù)將按順序執(zhí)行。若任務(wù)時(shí)間沖突，schedule不會(huì)主動(dòng)處理，而是按任務(wù)添加順序依次執(zhí)行。
import schedule
import time
from datetime import datetime
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
def job():
"""定時(shí)執(zhí)行的任務(wù)函數(shù)"""
print(f"{get_now_time()} I'm working...")
# 基于當(dāng)前時(shí)間，設(shè)置各種定時(shí)任務(wù)
# 基于當(dāng)前時(shí)間，每分鐘的16秒執(zhí)行任務(wù)
schedule.every().minutes.at(":16").do(job)
# 基于當(dāng)前時(shí)間，每小時(shí)的第23分鐘00秒執(zhí)行任務(wù)
schedule.every().hours.at(":23").do(job)
# 基于當(dāng)前時(shí)間，每5小時(shí)的第20分30秒執(zhí)行任務(wù)
schedule.every(5).hours.at("20:30").do(job)
# 基于當(dāng)前時(shí)間，每天上午10:30:00執(zhí)行任務(wù)
schedule.every().days.at("10:30").do(job)
# 基于當(dāng)前時(shí)間，每天上午10:30:42執(zhí)行任務(wù)
schedule.every().days.at("10:30:42").do(job)
# 基于當(dāng)前時(shí)間，每天上午12:42（阿姆斯特丹時(shí)區(qū)）執(zhí)行任務(wù)
# 需要安裝pip install pytz
schedule.every().days.at("12:42", "Europe/Amsterdam").do(job)
# 基于當(dāng)前時(shí)間，每周三下午1點(diǎn)15分執(zhí)行任務(wù)
schedule.every().wednesday.at("13:15").do(job)
# 任務(wù)調(diào)度主循環(huán)
# 持續(xù)檢查是否有待執(zhí)行的任務(wù)
while True:
schedule.run_pending()
time.sleep(1)
1.2 進(jìn)階使用
1.2.1 調(diào)用程序管理
裝飾器調(diào)用
可以通過使用@repeat裝飾器來調(diào)度函數(shù)。傳遞給它一個(gè)時(shí)間間隔，但省略do函數(shù)：
from schedule import every, repeat, run_pending
import time
from datetime import datetime
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
@repeat(every(2).seconds)
def job():
print(f"{get_now_time()} I'm working...")
while True:
run_pending()
time.sleep(1)
參數(shù)傳遞
在調(diào)用時(shí)，可以通過do函數(shù)傳遞額外的參數(shù)給任務(wù)函數(shù)：
import schedule
import time
from schedule import every, repeat, run_pending
from datetime import datetime
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
def job(name, message):
print(f"{get_now_time()} {message} {name}")
# 傳遞name和message參數(shù)
schedule.every(2).seconds.do(job, name='world!', message='hello')
@repeat(every().seconds, "code", "good")
def hello(name, message):
print(message, name)
while True:
run_pending()
time.sleep(1)
任務(wù)取消
若要從調(diào)度器中移除任務(wù)，可使用schedule.cancel_job(job)方法。
import schedule
def some_task():
print('Hello world')
job = schedule.every().days.at('12:30').do(some_task)
schedule.cancel_job(job)
任務(wù)移除
從任務(wù)調(diào)用的函數(shù)中返回schedule.CancelJob可以將其從調(diào)度器中移除，注意以下代碼由于有while函數(shù)，移除后代碼并不會(huì)退出：
import schedule
import time
def job_that_executes_once(name):
print(f"hello {name}")
return schedule.CancelJob
schedule.every().minutes.at(':30').do(job_that_executes_once,name="job1")
while True:
schedule.run_pending()
time.sleep(1)
任務(wù)批管理
以下代碼展示了獲取所有運(yùn)行的任務(wù)，同時(shí)一次性清除所有任務(wù)：
# 導(dǎo)入 schedule 庫用于創(chuàng)建和管理定時(shí)任務(wù)
import schedule
# 定義任務(wù)函數(shù)，接收一個(gè)名字參數(shù)并打印問候語
def greet(name):
print('Hello {}'.format(name))
job1 = schedule.every().second.do(greet, name='job1')
schedule.every().second.do(greet, name='job2')
# 獲取當(dāng)前所有已注冊(cè)的定時(shí)任務(wù)
all_jobs = schedule.get_jobs()
# 打印任務(wù)列表（包含任務(wù)類型、執(zhí)行周期、函數(shù)名和參數(shù)等信息）
print(all_jobs)
schedule.every().second.do(greet, name='job3')
# 取消job1任務(wù)（通過之前保存的任務(wù)對(duì)象引用）
# 取消后該任務(wù)將不再執(zhí)行
schedule.cancel_job(job1)
# 重新獲取當(dāng)前所有已注冊(cè)的定時(shí)任務(wù)
all_jobs = schedule.get_jobs()
print(all_jobs)
# 清除所有已注冊(cè)的定時(shí)任務(wù)
schedule.clear()
標(biāo)簽管理
以下代碼展示了如何為任務(wù)設(shè)置標(biāo)簽，并基于標(biāo)簽挑選和管理任務(wù)：
import schedule
import time
def greet(name):
print(f"Hello {name}")
# 創(chuàng)建帶標(biāo)簽的定時(shí)任務(wù)
schedule.every().days.do(greet, 'Andrea').tag('daily-tasks', 'friend')
schedule.every().hours.do(greet, 'John').tag('hourly-tasks', 'friend')
schedule.every().hours.do(greet, 'Monica').tag('hourly-tasks', 'customer')
schedule.every().days.do(greet, 'Derek').tag('daily-tasks', 'guest')
# 獲取特定標(biāo)簽的任務(wù)
friends = schedule.get_jobs('friend')
print("所有帶有friend標(biāo)簽的任務(wù):")
# 取消帶有daily-tasks標(biāo)簽的任務(wù)
schedule.clear('daily-tasks')
for job in friends:
print(f"- {job}")
# 運(yùn)行所有待執(zhí)行的任務(wù)
while True:
schedule.run_pending()
time.sleep(1)
1.2.2 調(diào)用時(shí)間管理
隨機(jī)時(shí)間
以下代碼展示了按隨機(jī)間隔運(yùn)行任務(wù)的功能：
import schedule
import time
from datetime import datetime
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
def my_job():
print(f"{get_now_time()} hello")
# 在1秒到5秒這個(gè)閉區(qū)間內(nèi)，隨機(jī)選擇一個(gè)間隔時(shí)間，周期性地執(zhí)行指定的任務(wù)函數(shù)
schedule.every(1).to(5).seconds.do(my_job)
while True:
schedule.run_pending()
time.sleep(1)
截至?xí)r間
以下代碼展示如何調(diào)用until函數(shù)設(shè)置任務(wù)的截止時(shí)間，任務(wù)在截止時(shí)間之后將不再運(yùn)行。
import schedule
from datetime import datetime, timedelta, time
from datetime import datetime
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
def job():
print(f"{get_now_time()} hello")
# 在今天22:30前，每隔1小時(shí)執(zhí)行一次任務(wù)
schedule.every(1).hours.until("22:30").do(job)
# 在2030-01-01 22:33前，每隔1小時(shí)執(zhí)行一次任務(wù)
schedule.every(1).hours.until("2030-01-01 22:33").do(job)
# 在接下來的8小時(shí)內(nèi)，每隔1小時(shí)執(zhí)行一次任務(wù)
schedule.every(1).hours.until(timedelta(hours=8)).do(job)
# 在今天22:33:42前，每隔1小時(shí)執(zhí)行一次任務(wù)
schedule.every(1).hours.until(time(22, 33, 42)).do(job)
# 在2027-05-17 11:36:20前，每隔5秒執(zhí)行一次任務(wù)
schedule.every(5).seconds.until(datetime(2027, 5, 17, 11, 36, 20)).do(job)
# 主循環(huán)：持續(xù)檢查并執(zhí)行待處理的任務(wù)
while True:
schedule.run_pending()
獲取執(zhí)行時(shí)間
使用schedule.idle_seconds()來獲取距離下次任務(wù)計(jì)劃執(zhí)行的剩余秒數(shù)。如果下次計(jì)劃執(zhí)行的任務(wù)本應(yīng)在過去執(zhí)行，返回值為負(fù)數(shù)。若沒有計(jì)劃任務(wù)，則返回None。示例如下：
import schedule
import time
def job():
print('你好')
schedule.every(2).minutes.do(job)
while 1:
# n為距離下次執(zhí)行任務(wù)的剩余秒數(shù)
n = schedule.idle_seconds()
print(n)
if n is None:
# 沒有更多任務(wù)
break
elif n > 0:
# 精確睡眠相應(yīng)的時(shí)間
time.sleep(n)
schedule.run_pending()
1.2.3 調(diào)用運(yùn)行方式管理
任務(wù)全部運(yùn)行
以下代碼展示了通過run_all()忽略任務(wù)預(yù)設(shè)的時(shí)間安排，立即執(zhí)行所有已定義的任務(wù)。先定義的任務(wù)先執(zhí)行。示例如下：
import schedule
from datetime import datetime
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
def job_1():
print(f"{get_now_time()} job1")
def job_2():
print(f"{get_now_time()} job2")
schedule.every().monday.at("12:40").do(job_1)
schedule.every().tuesday.at("16:40").do(job_2)
# 立即運(yùn)行所有任務(wù)一次
schedule.run_all()
# 立即運(yùn)行所有任務(wù)，每個(gè)任務(wù)運(yùn)行間隔3秒
schedule.run_all(delay_seconds=3)
后臺(tái)運(yùn)行
默認(rèn)情況下，無法在后臺(tái)運(yùn)行調(diào)度程序。不過，通過創(chuàng)建一個(gè)線程，利用該線程在不阻塞主線程的情況下運(yùn)行任務(wù)。以下是實(shí)現(xiàn)這一操作的示例：
import threading
import time
import schedule
def run_continuously(interval=1):
"""
創(chuàng)建一個(gè)后臺(tái)線程持續(xù)運(yùn)行調(diào)度器
"""
# 創(chuàng)建事件對(duì)象用于控制線程停止
# Event是事件類，線程間通信的簡(jiǎn)單機(jī)制，有"set"和"clear"兩種狀態(tài)
# 初始狀態(tài)為"clear"，通過cease_continuous_run.is_set()判斷是否為set
cease_continuous_run = threading.Event()
class ScheduleThread(threading.Thread):
@classmethod
def run(cls):
# 在線程啟動(dòng)后循環(huán)執(zhí)行
# 只要任務(wù)狀態(tài)不是set，就運(yùn)行任務(wù)
while not cease_continuous_run.is_set():
# 檢查并執(zhí)行所有待執(zhí)行的定時(shí)任務(wù)
schedule.run_pending()
# 休眠指定間隔時(shí)間
time.sleep(interval)
# 創(chuàng)建并啟動(dòng)調(diào)度線程
continuous_thread = ScheduleThread()
continuous_thread.start()
# 返回事件對(duì)象用于后續(xù)停止線程
return cease_continuous_run
def background_job():
"""定時(shí)執(zhí)行的后臺(tái)任務(wù)"""
print('Hello from the background thread')
# 設(shè)置定時(shí)任務(wù)：每秒執(zhí)行一次background_job函數(shù)
schedule.every().second.do(background_job)
# 啟動(dòng)后臺(tái)調(diào)度線程并獲取停止控制器
stop_run_continuously = run_continuously()
# 主線程繼續(xù)執(zhí)行其他任務(wù)
print("主線程繼續(xù)執(zhí)行中...")
time.sleep(5)
# 停止后臺(tái)調(diào)度線程
# 將事件對(duì)象的內(nèi)部標(biāo)志設(shè)置為set
stop_run_continuously.set()
print("后臺(tái)線程已停止")
多任務(wù)同時(shí)執(zhí)行
默認(rèn)情況下，schedule任務(wù)調(diào)度工具會(huì)按順序逐個(gè)執(zhí)行所有任務(wù)。以10秒內(nèi)執(zhí)行30個(gè)任務(wù)為例，從日志中可以看到，這些任務(wù)會(huì)在這10秒內(nèi)以串行方式依次執(zhí)行，如同排隊(duì)等候處理，而非同時(shí)運(yùn)行。這種設(shè)計(jì)的核心目的是避免多個(gè)任務(wù)搶占資源或產(chǎn)生相互干擾，從而確保任務(wù)執(zhí)行的穩(wěn)定性和可靠性。
若需要實(shí)現(xiàn)多個(gè)任務(wù)并行運(yùn)行，可通過為每個(gè)任務(wù)分配獨(dú)立線程的方式達(dá)成，并通過統(tǒng)一隊(duì)列進(jìn)行調(diào)用。如下所示：
import time
import threading
import schedule
import queue
from datetime import datetime
# 獲取當(dāng)前時(shí)間并格式化為字符串
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
# 定義任務(wù)函數(shù)，打印當(dāng)前時(shí)間和任務(wù)名稱
def job(name):
print(f"{get_now_time()} {name}")
# 工作線程主函數(shù)，負(fù)責(zé)從隊(duì)列中獲取并執(zhí)行任務(wù)
def worker_main():
while True:
joo_func,name = jobqueue.get()
joo_func(name)
jobqueue.task_done()
# 創(chuàng)建任務(wù)隊(duì)列
jobqueue = queue.Queue()
# 調(diào)度多個(gè)相同間隔的任務(wù)，將任務(wù)放入隊(duì)列
schedule.every(5).seconds.do(jobqueue.put, [job,"job1"])
schedule.every(5).seconds.do(jobqueue.put, [job,"job2"])
schedule.every(5).seconds.do(jobqueue.put, [job,"job3"])
schedule.every(5).seconds.do(jobqueue.put, [job,"job4"])
schedule.every(5).seconds.do(jobqueue.put, [job,"job5"])
# 啟動(dòng)工作線程，對(duì)對(duì)任務(wù)隊(duì)列進(jìn)行處理
worker_thread = threading.Thread(target=worker_main)
worker_thread.start()
while True:
schedule.run_pending()
time.sleep(1)
如果不需要隊(duì)列統(tǒng)一調(diào)用，代碼如下：
import time
import threading
import schedule
import queue
from datetime import datetime
# 獲取當(dāng)前時(shí)間并格式化為字符串
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
# 定義任務(wù)函數(shù)，打印當(dāng)前時(shí)間和任務(wù)名稱
def job(name):
print(f"{get_now_time()} {name}")
def run_threaded(job_func,name):
job_thread = threading.Thread(target=job_func,args=(name,))
job_thread.start()
schedule.every(5).seconds.do(run_threaded, job, 'job1')
schedule.every(5).seconds.do(run_threaded, job, 'job2')
schedule.every(5).seconds.do(run_threaded, job, 'job3')
schedule.every(5).seconds.do(run_threaded, job, 'job4')
schedule.every(5).seconds.do(run_threaded, job, 'job5')
while True:
schedule.run_pending()
time.sleep(1)
異常處理
調(diào)度程序不會(huì)捕獲任務(wù)執(zhí)行過程中發(fā)生的異常，并將異常傳遞給調(diào)用函數(shù)，可能直接崩潰程序：
import schedule
import time
def bad_task():
return 1 / 0
schedule.every(1).minutes.do(bad_task)
while True:
schedule.run_pending()
time.sleep(1)
如果需要防范此類異常，可以按如下方式用裝飾器封裝任務(wù)函數(shù)：
import schedule
import time
import functools # 導(dǎo)入函數(shù)工具庫
# 定義一個(gè)捕獲異常的裝飾器，用于包裝定時(shí)任務(wù)
def catch_exceptions(cancel_on_failure=False):
def catch_exceptions_decorator(job_func):
@functools.wraps(job_func) # 保留被裝飾函數(shù)的元信息
def wrapper(*args, **kwargs):
try:
return job_func(*args, **kwargs) # 執(zhí)行原函數(shù)
except:
import traceback
print(traceback.format_exc()) # 打印完整的異常堆棧信息
if cancel_on_failure: # 如果設(shè)置了失敗后取消任務(wù)
return schedule.CancelJob # 返回取消任務(wù)的標(biāo)志
return wrapper
return catch_exceptions_decorator
# 使用裝飾器包裝任務(wù)函數(shù)，設(shè)置失敗后自動(dòng)取消
@catch_exceptions(cancel_on_failure=True)
def bad_task():
return 1 / 0
schedule.every(1).minutes.do(bad_task)
# 主循環(huán)：持續(xù)檢查并執(zhí)行待處理的任務(wù)
while True:
schedule.run_pending()
time.sleep(1)
日志管理
通過logging庫設(shè)置名為schedule的日志記錄器并設(shè)置為DEBUG級(jí)別，使其能夠捕獲并處理schedule庫內(nèi)部產(chǎn)生的所有日志信息：
import schedule
import logging
# 配置基本日志設(shè)置
logging.basicConfig()
# 獲取名為'schedule'的日志記錄器
# 由于schedule庫在內(nèi)部使用相同的名稱（'schedule'）記錄自己的日志
# 因此這個(gè)記錄器可以捕獲并處理schedule模塊產(chǎn)生的所有日志信息
schedule_logger = logging.getLogger('schedule')
# 設(shè)置日志級(jí)別為DEBUG，以便記錄詳細(xì)的調(diào)試信息
schedule_logger.setLevel(level=logging.DEBUG)
def job():
print("Hello, Logs") # 打印信息到標(biāo)準(zhǔn)輸出和log
schedule.every().second.do(job)
# 立即運(yùn)行所有已安排的任務(wù)（僅執(zhí)行一次）
schedule.run_all()
# 清除所有已安排的任務(wù)
schedule.clear()
如果若想為任務(wù)添加可復(fù)用的日志記錄功能，最簡(jiǎn)便的方法是實(shí)現(xiàn)一個(gè)處理日志的裝飾器：
import functools
import time
import schedule
# 這個(gè)裝飾器可用于任何任務(wù)函數(shù)，用于記錄每次任務(wù)的執(zhí)行時(shí)間
def print_elapsed_time(func):
@functools.wraps(func) # 讓被裝飾函數(shù)的名稱、文檔字符串等屬性保持不變。
def wrapper(*args, **kwargs):
# 記錄任務(wù)開始時(shí)間戳
start_timestamp = time.time()
print(f'LOG: 正在運(yùn)行任務(wù) "{func.__name__}"')
# 執(zhí)行實(shí)際任務(wù)
result = func(*args, **kwargs)
# 計(jì)算并打印任務(wù)執(zhí)行耗時(shí)
print(f'LOG: 任務(wù) "{func.__name__}" 已完成，耗時(shí) {time.time() - start_timestamp:.1f} 秒')
return result
return wrapper
# 應(yīng)用裝飾器，自動(dòng)記錄該任務(wù)的執(zhí)行時(shí)間
@print_elapsed_time
def job():
print('Hello, Logs')
# 模擬耗時(shí)操作
time.sleep(2)
schedule.every().second.do(job)
# 立即運(yùn)行所有已注冊(cè)的任務(wù)一次
schedule.run_all()
多調(diào)度程序運(yùn)行
從一個(gè)調(diào)度程序里運(yùn)行多少個(gè)任務(wù)都可以。不過要是調(diào)度程序規(guī)模比較大，可能需要用多個(gè)調(diào)度程序來管理。如下所示：
import time
import schedule
from datetime import datetime
# 獲取當(dāng)前時(shí)間并格式化為字符串
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
def fooJob(caller):
print(f"{get_now_time()} Foo called by {caller}")
def barJob(caller):
print(f"{get_now_time()} Bar called by {caller}")
scheduler1 = schedule.Scheduler()
scheduler1.every().hour.do(fooJob, caller="scheduler1")
scheduler1.every().hour.do(barJob, caller="scheduler1")
scheduler2 = schedule.Scheduler()
scheduler2.every().second.do(fooJob, caller="scheduler2")
scheduler2.every().second.do(barJob, caller="scheduler2")
# 主循環(huán)，使程序持續(xù)運(yùn)行，不斷檢查并執(zhí)行待處理的任務(wù)
while True:
# 檢查scheduler1中是否有待執(zhí)行的任務(wù)，若有則執(zhí)行
scheduler1.run_pending()
# 檢查scheduler2中是否有待執(zhí)行的任務(wù)，若有則執(zhí)行
scheduler2.run_pending()
time.sleep(1)以上就是Python中輕量級(jí)定時(shí)任務(wù)調(diào)度庫schedule使用指南的詳細(xì)內(nèi)容，更多關(guān)于Python schedule庫使用的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
Python實(shí)現(xiàn)四舍五入的兩個(gè)方法總結(jié)
這篇文章主要介紹了python中實(shí)現(xiàn)四舍五入的兩種方法，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-09-09
python實(shí)現(xiàn)簡(jiǎn)易SSL的項(xiàng)目實(shí)踐
本文主要介紹了python實(shí)現(xiàn)簡(jiǎn)易SSL的項(xiàng)目實(shí)踐,包括CA.py、server.py和client.py三個(gè)模塊,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2025-02-02
Python基礎(chǔ)必備之語法結(jié)構(gòu)詳解
Python語法定義了用于在 Python 編程中創(chuàng)建句子的所有規(guī)則集。如果想更深入地研究 Python 詞法結(jié)構(gòu)，需要了解構(gòu)成語句的句法元素，即構(gòu)成 Python 程序的基本單元，涵蓋控制結(jié)構(gòu)，在不同代碼組之間引導(dǎo)程序流的構(gòu)造，快跟隨小編一起學(xué)習(xí)一下吧2022-04-04
Python 中的參數(shù)傳遞、返回值、淺拷貝、深拷貝
這篇文章主要介紹了Python 中的參數(shù)傳遞、返回值、淺拷貝、深拷貝,非常不錯(cuò)，具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2019-06-06
Python全局變量關(guān)鍵字global的簡(jiǎn)單使用
python中g(shù)lobal關(guān)鍵字主要作用是聲明變量的作用域,下面這篇文章主要給大家介紹了關(guān)于Python全局變量關(guān)鍵字global的簡(jiǎn)單使用,文中通過實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下2022-06-06
這篇文章主要為大家詳細(xì)介紹了python的concat等多種用法，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-11-11
這篇文章主要為大家介紹了python數(shù)字圖像處理像素的訪問與裁剪示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-06-06
這篇文章主要介紹了在Python中使用PIL模塊處理圖像的教程,PIL模塊在Python編程中也是十分常用的模塊,示例代碼基于Python2.x版本,需要的朋友可以參考下2015-04-04
這篇文章主要介紹了python打包成so文件過程解析,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-09-09
這篇文章主要介紹了Python中的main函數(shù)與import用法，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-11-11