欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Python中輕量級定時任務(wù)調(diào)度庫schedule使用指南

 更新時間:2025年05月21日 08:11:25   作者:落痕的寒假  
schedule是一款專為簡化定時任務(wù)調(diào)度而設(shè)計的Python庫,它通過直觀的語法降低了周期性任務(wù)的實現(xiàn)門檻,下面小編來和大家詳細講講它的具體使用吧

schedule是一款專為簡化定時任務(wù)調(diào)度而設(shè)計的Python庫,它通過直觀的語法降低了周期性任務(wù)的實現(xiàn)門檻。作為進程內(nèi)調(diào)度器,它無需額外守護進程,輕量且無外部依賴,適合快速搭建自動化任務(wù)。不過,該庫在功能完整性上有所取舍,目前暫不支持斷點續(xù)傳、亞秒級精度控制以及多任務(wù)并行執(zhí)行等復(fù)雜場景。

schedule庫的官方倉庫地址見:schedule,schedule庫的官方文檔見:schedule-doc

schedule庫支持在Python 3.7及以上版本的環(huán)境中運行,schedule庫的安裝命令如下:

pip install schedule

1 使用入門

1.1 基礎(chǔ)使用

1.1.1 相對調(diào)用

schedule庫通過創(chuàng)建調(diào)度器,設(shè)置時間單位,注冊待調(diào)用函數(shù),返回任務(wù)對象實現(xiàn)任務(wù)周期調(diào)用。例如:

# 每3秒執(zhí)行一次任務(wù)
schedule.every(3).seconds.do(job)

這種調(diào)用方式是相對調(diào)用方式,任務(wù)執(zhí)行間隔是相對于當前時間點計算的。例如17:59:01創(chuàng)建調(diào)度器,下一次執(zhí)行時間為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():
    """定時執(zhí)行的任務(wù)函數(shù)"""
    print(f"{get_now_time()} I'm working...") 

print(f"調(diào)度器創(chuàng)建時間:{get_now_time()}")
# 調(diào)度器返回job對象
schedule_job = schedule.every(3).seconds.do(job)

print(f"調(diào)度器下一次運行時間:{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就是沒運行
    print(f"調(diào)度器上一次運行時間:{schedule_job.last_run}")
    print(f"調(diào)度器下一次運行時間:{schedule_job.next_run}")

其他相對時間間隔調(diào)用代碼如下:

# 每3分鐘執(zhí)行一次任務(wù)
schedule.every(3).minutes.do(job)
# 每小時執(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 絕對調(diào)用

schedule可以在相對調(diào)用的基礎(chǔ)上結(jié)合at函數(shù)實現(xiàn)絕對時間的調(diào)度。例如:

schedule.every(3).hours.at("11:16").do(job)

上述代碼可拆解為:

job1 = schedule.every(3).hours
job2 = job1.at("11:16").do(job)

此處的job1基于當前調(diào)用器創(chuàng)建時間(例如18:34:54),以小時為間隔粒度進行設(shè)定,即每間隔3小時執(zhí)行一次任務(wù),因此下一次任務(wù)執(zhí)行時間為21:34:54。而at()函數(shù)的作用是在job1設(shè)定的時間粒度和間隔范圍內(nèi),具體指定分鐘和秒。以job2中的at("11:16")為例,這里的11:16代表分鐘和秒。它會在18:34:5421:34:54的時間區(qū)間內(nèi),定位離21:34:54最近的1116秒時刻,即21:11:16

若按每分鐘調(diào)用一次任務(wù),可通過at指定固定執(zhí)行秒數(shù)。例如,當創(chuàng)建時間為18:34:54時,下一次執(zhí)行時間為18:35:16,對應(yīng)代碼如下:

schedule_job = schedule.every().minutes.at(":16").do(job)

at函數(shù)的輸入范圍由設(shè)定的時間粒度決定,且僅支持日級、時級、分級三類時間粒度,具體子粒度支持規(guī)則如下:

  • 每日任務(wù):支持HH:MM:SS(時分秒)和HH:MM兩種格式(HH:MM默認補全為HH:MM:00);
  • 每小時任務(wù):支持MM:SS(分秒)和:MM兩種格式(:MM默認補全為MM:00);
  • 每分鐘任務(wù):僅支持:SS(秒)格式。

其他絕對時間間隔調(diào)用代碼如下:

# 基于當前時間,每分鐘的16秒執(zhí)行任務(wù)
schedule.every().minutes.at(":16").do(job)
# 基于當前時間,每小時的第23分鐘00秒執(zhí)行任務(wù)
schedule.every().hours.at(":23").do(job)
# 基于當前時間,每5小時的第20分30秒執(zhí)行任務(wù)
schedule.every(5).hours.at("20:30").do(job)
# 基于當前時間,每天上午10:30:00執(zhí)行任務(wù)
schedule.every().days.at("10:30").do(job)
# 基于當前時間,每天上午10:30:42執(zhí)行任務(wù)
schedule.every().days.at("10:30:42").do(job)
# 基于當前時間,每天上午12:42(阿姆斯特丹時區(qū))執(zhí)行任務(wù)
schedule.every().days.at("12:42", "Europe/Amsterdam").do(job)
# 基于當前時間,每周三下午1點15分執(zhí)行任務(wù)
schedule.every().wednesday.at("13:15").do(job)

以下示例代碼演示了多任務(wù)調(diào)度場景。作為輕量級任務(wù)調(diào)度庫,schedule會維護任務(wù)列表,調(diào)用schedule.run_pending()時檢查所有任務(wù)觸發(fā)條件,滿足條件的任務(wù)將按順序執(zhí)行。若任務(wù)時間沖突,schedule不會主動處理,而是按任務(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():
    """定時執(zhí)行的任務(wù)函數(shù)"""
    print(f"{get_now_time()} I'm working...") 

# 基于當前時間,設(shè)置各種定時任務(wù)
# 基于當前時間,每分鐘的16秒執(zhí)行任務(wù)
schedule.every().minutes.at(":16").do(job)
# 基于當前時間,每小時的第23分鐘00秒執(zhí)行任務(wù)
schedule.every().hours.at(":23").do(job)
# 基于當前時間,每5小時的第20分30秒執(zhí)行任務(wù)
schedule.every(5).hours.at("20:30").do(job)
# 基于當前時間,每天上午10:30:00執(zhí)行任務(wù)
schedule.every().days.at("10:30").do(job)
# 基于當前時間,每天上午10:30:42執(zhí)行任務(wù)
schedule.every().days.at("10:30:42").do(job)
# 基于當前時間,每天上午12:42(阿姆斯特丹時區(qū))執(zhí)行任務(wù)
# 需要安裝pip install pytz
schedule.every().days.at("12:42", "Europe/Amsterdam").do(job)
# 基于當前時間,每周三下午1點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 進階使用

1.2.1 調(diào)用程序管理

裝飾器調(diào)用

可以通過使用@repeat裝飾器來調(diào)度函數(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)用時,可以通過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ù),移除后代碼并不會退出:

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ù)批管理

以下代碼展示了獲取所有運行的任務(wù),同時一次性清除所有任務(wù):

# 導(dǎo)入 schedule 庫用于創(chuàng)建和管理定時任務(wù)
import schedule

# 定義任務(wù)函數(shù),接收一個名字參數(shù)并打印問候語
def greet(name):
    print('Hello {}'.format(name))

job1 = schedule.every().second.do(greet, name='job1')
schedule.every().second.do(greet, name='job2')

# 獲取當前所有已注冊的定時任務(wù)
all_jobs = schedule.get_jobs()
# 打印任務(wù)列表(包含任務(wù)類型、執(zhí)行周期、函數(shù)名和參數(shù)等信息)
print(all_jobs)

schedule.every().second.do(greet, name='job3')

# 取消job1任務(wù)(通過之前保存的任務(wù)對象引用)
# 取消后該任務(wù)將不再執(zhí)行
schedule.cancel_job(job1)

# 重新獲取當前所有已注冊的定時任務(wù)
all_jobs = schedule.get_jobs()
print(all_jobs)

# 清除所有已注冊的定時任務(wù)
schedule.clear()

標簽管理

以下代碼展示了如何為任務(wù)設(shè)置標簽,并基于標簽挑選和管理任務(wù):

import schedule
import time

def greet(name):
    print(f"Hello {name}")

# 創(chuàng)建帶標簽的定時任務(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')

# 獲取特定標簽的任務(wù)
friends = schedule.get_jobs('friend')
print("所有帶有friend標簽的任務(wù):")
# 取消帶有daily-tasks標簽的任務(wù)
schedule.clear('daily-tasks')

for job in friends:
    print(f"- {job}")

# 運行所有待執(zhí)行的任務(wù)
while True:
    schedule.run_pending()
    time.sleep(1)

1.2.2 調(diào)用時間管理

隨機時間

以下代碼展示了按隨機間隔運行任務(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秒這個閉區(qū)間內(nèi),隨機選擇一個間隔時間,周期性地執(zhí)行指定的任務(wù)函數(shù)
schedule.every(1).to(5).seconds.do(my_job)

while True:
    schedule.run_pending()
    time.sleep(1)

截至時間

以下代碼展示如何調(diào)用until函數(shù)設(shè)置任務(wù)的截止時間,任務(wù)在截止時間之后將不再運行。

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小時執(zhí)行一次任務(wù)
schedule.every(1).hours.until("22:30").do(job)

# 在2030-01-01 22:33前,每隔1小時執(zhí)行一次任務(wù)
schedule.every(1).hours.until("2030-01-01 22:33").do(job)

# 在接下來的8小時內(nèi),每隔1小時執(zhí)行一次任務(wù)
schedule.every(1).hours.until(timedelta(hours=8)).do(job)

# 在今天22:33:42前,每隔1小時執(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í)行時間

使用schedule.idle_seconds()來獲取距離下次任務(wù)計劃執(zhí)行的剩余秒數(shù)。如果下次計劃執(zhí)行的任務(wù)本應(yīng)在過去執(zhí)行,返回值為負數(shù)。若沒有計劃任務(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)的時間
        time.sleep(n)
    schedule.run_pending()

1.2.3 調(diào)用運行方式管理

任務(wù)全部運行

以下代碼展示了通過run_all()忽略任務(wù)預(yù)設(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)

# 立即運行所有任務(wù)一次
schedule.run_all()

# 立即運行所有任務(wù),每個任務(wù)運行間隔3秒
schedule.run_all(delay_seconds=3)

后臺運行

默認情況下,無法在后臺運行調(diào)度程序。不過,通過創(chuàng)建一個線程,利用該線程在不阻塞主線程的情況下運行任務(wù)。以下是實現(xiàn)這一操作的示例:

import threading
import time

import schedule

def run_continuously(interval=1):
    """
    創(chuàng)建一個后臺線程持續(xù)運行調(diào)度器
    """
    # 創(chuàng)建事件對象用于控制線程停止
    # Event是事件類,線程間通信的簡單機制,有"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):
            # 在線程啟動后循環(huán)執(zhí)行
            # 只要任務(wù)狀態(tài)不是set,就運行任務(wù)
            while not cease_continuous_run.is_set():
                # 檢查并執(zhí)行所有待執(zhí)行的定時任務(wù)
                schedule.run_pending()
                # 休眠指定間隔時間
                time.sleep(interval)

    # 創(chuàng)建并啟動調(diào)度線程
    continuous_thread = ScheduleThread()
    continuous_thread.start()
    
    # 返回事件對象用于后續(xù)停止線程
    return cease_continuous_run

def background_job():
    """定時執(zhí)行的后臺任務(wù)"""
    print('Hello from the background thread')

# 設(shè)置定時任務(wù):每秒執(zhí)行一次background_job函數(shù)
schedule.every().second.do(background_job)

# 啟動后臺調(diào)度線程并獲取停止控制器
stop_run_continuously = run_continuously()

# 主線程繼續(xù)執(zhí)行其他任務(wù)
print("主線程繼續(xù)執(zhí)行中...")
time.sleep(5)

# 停止后臺調(diào)度線程
# 將事件對象的內(nèi)部標志設(shè)置為set
stop_run_continuously.set()
print("后臺線程已停止")

多任務(wù)同時執(zhí)行

默認情況下,schedule任務(wù)調(diào)度工具會按順序逐個執(zhí)行所有任務(wù)。以10秒內(nèi)執(zhí)行30個任務(wù)為例,從日志中可以看到,這些任務(wù)會在這10秒內(nèi)以串行方式依次執(zhí)行,如同排隊等候處理,而非同時運行。這種設(shè)計的核心目的是避免多個任務(wù)搶占資源或產(chǎn)生相互干擾,從而確保任務(wù)執(zhí)行的穩(wěn)定性和可靠性。

若需要實現(xiàn)多個任務(wù)并行運行,可通過為每個任務(wù)分配獨立線程的方式達成,并通過統(tǒng)一隊列進行調(diào)用。如下所示:

import time
import threading
import schedule
import queue
from datetime import datetime

# 獲取當前時間并格式化為字符串
def get_now_time():
    now = datetime.now()
    now = now.strftime("%Y-%m-%d %H:%M:%S")
    return now

# 定義任務(wù)函數(shù),打印當前時間和任務(wù)名稱
def job(name):
    print(f"{get_now_time()} {name}")  

# 工作線程主函數(shù),負責從隊列中獲取并執(zhí)行任務(wù)
def worker_main():
    while True:
        joo_func,name = jobqueue.get()
        joo_func(name)
        jobqueue.task_done()

# 創(chuàng)建任務(wù)隊列
jobqueue = queue.Queue()

# 調(diào)度多個相同間隔的任務(wù),將任務(wù)放入隊列
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"])

# 啟動工作線程,對對任務(wù)隊列進行處理
worker_thread = threading.Thread(target=worker_main)
worker_thread.start()

while True:
    schedule.run_pending()
    time.sleep(1)

如果不需要隊列統(tǒng)一調(diào)用,代碼如下:

import time
import threading
import schedule
import queue
from datetime import datetime

# 獲取當前時間并格式化為字符串
def get_now_time():
    now = datetime.now()
    now = now.strftime("%Y-%m-%d %H:%M:%S")
    return now

# 定義任務(wù)函數(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)度程序不會捕獲任務(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ù)工具庫

# 定義一個捕獲異常的裝飾器,用于包裝定時任務(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ù)的標志
        return wrapper
    return catch_exceptions_decorator

# 使用裝飾器包裝任務(wù)函數(shù),設(shè)置失敗后自動取消
@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級別,使其能夠捕獲并處理schedule庫內(nèi)部產(chǎn)生的所有日志信息:

import schedule  
import logging  

# 配置基本日志設(shè)置
logging.basicConfig()
# 獲取名為'schedule'的日志記錄器
# 由于schedule庫在內(nèi)部使用相同的名稱('schedule')記錄自己的日志
# 因此這個記錄器可以捕獲并處理schedule模塊產(chǎn)生的所有日志信息
schedule_logger = logging.getLogger('schedule')
# 設(shè)置日志級別為DEBUG,以便記錄詳細的調(diào)試信息
schedule_logger.setLevel(level=logging.DEBUG)

def job():
    print("Hello, Logs")  # 打印信息到標準輸出和log

schedule.every().second.do(job)

# 立即運行所有已安排的任務(wù)(僅執(zhí)行一次)
schedule.run_all()

# 清除所有已安排的任務(wù)
schedule.clear()

如果若想為任務(wù)添加可復(fù)用的日志記錄功能,最簡便的方法是實現(xiàn)一個處理日志的裝飾器:

import functools
import time
import schedule

# 這個裝飾器可用于任何任務(wù)函數(shù),用于記錄每次任務(wù)的執(zhí)行時間
def print_elapsed_time(func):
    @functools.wraps(func) # 讓被裝飾函數(shù)的名稱、文檔字符串等屬性保持不變。
    def wrapper(*args, **kwargs):
        # 記錄任務(wù)開始時間戳
        start_timestamp = time.time()
        print(f'LOG: 正在運行任務(wù) "{func.__name__}"')
        
        # 執(zhí)行實際任務(wù)
        result = func(*args, **kwargs)
        
        # 計算并打印任務(wù)執(zhí)行耗時
        print(f'LOG: 任務(wù) "{func.__name__}" 已完成,耗時 {time.time() - start_timestamp:.1f} 秒')
        return result

    return wrapper

# 應(yīng)用裝飾器,自動記錄該任務(wù)的執(zhí)行時間
@print_elapsed_time
def job():
    print('Hello, Logs')
    # 模擬耗時操作
    time.sleep(2)

schedule.every().second.do(job)

# 立即運行所有已注冊的任務(wù)一次
schedule.run_all()

多調(diào)度程序運行

從一個調(diào)度程序里運行多少個任務(wù)都可以。不過要是調(diào)度程序規(guī)模比較大,可能需要用多個調(diào)度程序來管理。如下所示:

import time
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 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ù)運行,不斷檢查并執(zhí)行待處理的任務(wù)
while True:
    # 檢查scheduler1中是否有待執(zhí)行的任務(wù),若有則執(zhí)行
    scheduler1.run_pending()
    # 檢查scheduler2中是否有待執(zhí)行的任務(wù),若有則執(zhí)行
    scheduler2.run_pending()
    time.sleep(1)

以上就是Python中輕量級定時任務(wù)調(diào)度庫schedule使用指南的詳細內(nèi)容,更多關(guān)于Python schedule庫使用的資料請關(guān)注腳本之家其它相關(guān)文章!

相關(guān)文章

  • Python實現(xiàn)四舍五入的兩個方法總結(jié)

    Python實現(xiàn)四舍五入的兩個方法總結(jié)

    這篇文章主要介紹了python中實現(xiàn)四舍五入的兩種方法,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
    2022-09-09
  • python實現(xiàn)簡易SSL的項目實踐

    python實現(xiàn)簡易SSL的項目實踐

    本文主要介紹了python實現(xiàn)簡易SSL的項目實踐,包括CA.py、server.py和client.py三個模塊,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
    2025-02-02
  • Python基礎(chǔ)必備之語法結(jié)構(gòu)詳解

    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)造,快跟隨小編一起學習一下吧
    2022-04-04
  • Python 中的參數(shù)傳遞、返回值、淺拷貝、深拷貝

    Python 中的參數(shù)傳遞、返回值、淺拷貝、深拷貝

    這篇文章主要介紹了Python 中的參數(shù)傳遞、返回值、淺拷貝、深拷貝,非常不錯,具有一定的參考借鑒價值,需要的朋友可以參考下
    2019-06-06
  • Python全局變量關(guān)鍵字global的簡單使用

    Python全局變量關(guān)鍵字global的簡單使用

    python中g(shù)lobal關(guān)鍵字主要作用是聲明變量的作用域,下面這篇文章主要給大家介紹了關(guān)于Python全局變量關(guān)鍵字global的簡單使用,文中通過實例代碼介紹的非常詳細,需要的朋友可以參考下
    2022-06-06
  • python的concat等多種用法詳解

    python的concat等多種用法詳解

    這篇文章主要為大家詳細介紹了python的concat等多種用法,具有一定的參考價值,感興趣的小伙伴們可以參考一下
    2018-11-11
  • python數(shù)字圖像處理像素的訪問與裁剪示例

    python數(shù)字圖像處理像素的訪問與裁剪示例

    這篇文章主要為大家介紹了python數(shù)字圖像處理像素的訪問與裁剪示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪
    2022-06-06
  • 在Python中使用PIL模塊處理圖像的教程

    在Python中使用PIL模塊處理圖像的教程

    這篇文章主要介紹了在Python中使用PIL模塊處理圖像的教程,PIL模塊在Python編程中也是十分常用的模塊,示例代碼基于Python2.x版本,需要的朋友可以參考下
    2015-04-04
  • python打包成so文件過程解析

    python打包成so文件過程解析

    這篇文章主要介紹了python打包成so文件過程解析,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
    2019-09-09
  • Python中的main函數(shù)與import用法

    Python中的main函數(shù)與import用法

    這篇文章主要介紹了Python中的main函數(shù)與import用法,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2022-11-11

最新評論