Python中轻量级定时任务调度库schedule使用指南

目录

• 1 使用入门
• 1.1 基础使用
• 1.2 进阶使用

schedule是一款专为简化定时任务调度而设计的python库，它通过直观的语法降低了周期性任务的实现门槛。作为进程内调度器，它无需额外守护进程，轻量且无外部依赖，适合快速搭建自动化任务。不过，该库在功能完整性上有所取舍，目前暂不支持断点续传、亚秒级精度控制以及多任务并行执行等复杂场景。

schedule库的官方仓库地址见：schedule，schedule库的官方文档见：schedule-doc。

schedule库支持在Python 3.7及以上版本的环境中运行，schedule库的安装命令如下：

pip install schedule

1 使用入门

1.1 基础使用

1.1.1 相对调用

schedule库通过创建调度器，设置时间单位，注册待调用函数，返回任务对象实现任务周期调用。例如：

# 每3秒执行一次任务
schedule.every(3).seconds.do(job)

这种调用方式是相对调用方式，任务执行间隔是相对于当前时间点计算的。例如17:59:01创建调度器，下一次执行时间为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():
    """定时执行的任务函数"""
    print(f"{get_now_time()} I'm working...") 

print(f"调度器创建时间：{get_now_time()}")
# 调度器返回job对象
schedule_job = schedule.every(3).seconds.do(job)

print(f"调度器下一次运行时间：{schedule_job.next_run}")

# 任务调度主循环
# 持续检查是否有待执行的任务
while True:
    schedule.run_pending()  # 检查并执行待执行的任务
    time.sleep(1)          # 休眠1秒避免CPU占用过高
    # 为None就是没运行
    print(f"调度器上一次运行时间：{schedule_job.last_run}")
    print(f"调度器下一次运行时间：{schedule_job.next_run}")

其他相对时间间隔调用代码如下：

# 每3分钟执行一次任务
schedule.every(3).minutes.do(job)
# 每小时执行一次任务
schedule.every().hours.do(job)
# 每3天执行一次任务
schedule.every(3).days.do(job)
# 每3周执行一次任务
schedule.every(3).weeks.do(job)
# 每周一执行任务
schedule.every().monday.do(job)

1.1.2 绝对调用

schedule可以在相对调用的基础上结合at函数实现绝对时间的调度。例如：

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

上述代码可拆解为：

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

此处的job1基于当前调用器创建时间（例如18:34:54），以小时为间隔粒度进行设定，即每间隔3小时执行一次任务，因此下一次任务执行时间为21:34:54。而at()函数的作用是在job1设定的时间粒度和间隔范围内，具体指定分钟和秒。以job2中的at("11:16")为例，这里的11:16代表分钟和秒。它会在18:34:54至21:34:54的时间区间内，定位离21:34:54最近的11分16秒时刻，即21:11:16。

若按每分钟调用一次任务，可通过at指定固定执行秒数。例如，当创建时间为18:34:54时，下一次执行时间为18:35:16，对应代码如下：

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

at函数的输入范围由设定的时间粒度决定，且仅支持日级、时级、分级三类时间粒度，具体子粒度支持规则如下：

• 每日任务：支持HH:MM:SS（时分秒）和HH:MM两种格式（HH:MM默认补全为HH:MM:00）；
• 每小时任务：支持MM:SS（分秒）和:MM两种格式（:MM默认补全为MM:00）；
• 每分钟任务：仅支持:SS（秒）格式。

其他绝对时间间隔调用代码如下：

# 基于当前时间，每分钟的16秒执行任务
schedule.every().minutes.at(":16").do(job)
# 基于当前时间，每小时的第23分钟00秒执行任务
schedule.every().hours.at(":23").do(job)
# 基于当前时间，每5小时的第20分30秒执行任务
schedule.every(5).hours.at("20:30").do(job)
# 基于当前时间，每天上午10:30:00执行任务
schedule.every().days.at("10:30").do(job)
# 基于当前时间，每天上午10:30:42执行任务
schedule.every().days.at("10:30:42").do(job)
# 基于当前时间，每天上午12:42（阿姆斯特丹时区）执行任务
schedule.every().days.at("12:42", "Europe/Amsterdam").do(job)
# 基于当前时间，每周三下午1点15分执行任务
schedule.every().wednesday.at("13:15").do(job)

以下示例代码演示了多任务调度场景。作为轻量级任务调度库，schedule会维护任务列表，调用schedule.run_pending()时检查所有任务触发条件，满足条件的任务将按顺序执行。若任务时间冲突，schedule不会主动处理，而是按任务添加顺序依次执行。

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():
    """定时执行的任务函数"""
    print(f"{get_now_time()} I'm working...") 

# 基于当前时间，设置各种定时任务
# 基于当前时间，每分钟的16秒执行任务
schedule.every().minutes.at(":16").do(job)
# 基于当前时间，每小时的第23分钟00秒执行任务
schedule.every().hours.at(":23").do(job)
# 基于当前时间，每5小时的第20分30秒执行任务
schedule.every(5).hours.at("20:30").do(job)
# 基于当前时间，每天上午10:30:00执行任务
schedule.every().days.at("10:30").do(job)
# 基于当前时间，每天上午10:30:42执行任务
schedule.every().days.at("10:30:42").do(job)
# 基于当前时间，每天上午12:42（阿姆斯特丹时区）执行任务
# 需要安装pip install pytz
schedule.every().days.at("12:42", "Europe/Amsterdam").do(job)
# 基于当前时间，每周三下午1点15分执行任务
schedule.every().wednesday.at("13:15").do(job)

# 任务调度主循环
# 持续检查是否有待执行的任务
while True:
    schedule.run_pending()  
    time.sleep(1)          

1.2 进阶使用

1.2.1 调用程序管理

装饰器调用

可以通过使用@repeat装饰器来调度函数。传递给它一个时间间隔，但省略do函数：

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)

参数传递

在调用时，可以通过do函数传递额外的参数给任务函数：

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参数
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)

任务取消

若要从调度器中移除任务，可使用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)

任务移除

从任务调用的函数中返回schedule.CancelJob可以将其从调度器中移除，注意以下代码由于有while函数，移除后代码并不会退出：

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)

任务批管理

以下代码展示了获取所有运行的任务，同时一次性清除所有任务：

# 导入 schedule 库用于创建和管理定时任务
import schedule

# 定义任务函数，接收一个名字参数并打印问候语
def greet(name):
    print('Hello {}'.format(name))

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

# 获取当前所有已注册的定时任务
all_jobs = schedule.get_jobs()
# 打印任务列表（包含任务类型、执行周期、函数名和参数等信息）
print(all_jobs)

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

# 取消job1任务（通过之前保存的任务对象引用）
# 取消后该任务将不再执行
schedule.cancel_job(job1)

# 重新获取当前所有已注册的定时任务
all_jobs = schedule.get_jobs()
print(all_jobs)

# 清除所有已注册的定时任务
schedule.clear()

标签管理

以下代码展示了如何为任务设置标签，并基于标签挑选和管理任务：

import schedule
import time

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

# 创建带标签的定时任务
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')

# 获取特定标签的任务
friends = schedule.get_jobs('friend')
print("所有带有friend标签的任务:")
# 取消带有daily-tasks标签的任务
schedule.clear('daily-tasks')

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

# 运行所有待执行的任务
while True:
    schedule.run_pending()
    time.sleep(1)

1.2.2 调用时间管理

随机时间

以下代码展示了按随机间隔运行任务的功能：

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秒这个闭区间内，随机选择一个间隔时间，周期性地执行指定的任务函数
schedule.every(1).to(5).seconds.do(my_job)

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

截至时间

以下代码展示如何调用until函数设置任务的截止时间，任务在截止时间之后将不再运行。

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小时执行一次任务
schedule.every(1).hours.until("22:30").do(job)

# 在2030-01-01 22:33前，每隔1小时执行一次任务
schedule.every(1).hours.until("2030-01-01 22:33").do(job)

# 在接下来的8小时内，每隔1小时执行一次任务
schedule.every(1).hours.until(timedelta(hours=8)).do(job)

# 在今天22:33:42前，每隔1小时执行一次任务
schedule.every(1).hours.until(time(22, 33, 42)).do(job)

# 在2027-05-17 11:36:20前，每隔5秒执行http://www.devze.com一次任务
schedule.every(5).seconds.until(datetime(2027, 5, 17, 11, 36, 20)).do(job)

# 主循环：持续检查并执行待处理的任务
while True:
    schedule.run_pending()

获取执行时间

使用schedule.idle_seconds()来获取距离下次任务计划执行的剩余秒数。如果下次计划执行的任务本应在过去执行，返回值为负数。若没有计划任务，则返回None。示例如下：

import schedule
import time

def job():
    print('你好')

schedule.every(2).minutes.do(job)

while 1:
    # n为距离下次执行任务的剩余秒数
    n = schedule.idle_seconds()
    print(n)
    if n is None:
        # 没有更多任务
        break
    elif n > 0:
        # 精确睡眠相应的时间
        time.sleep(n)
    schedule.run_pending()

1.2.3 调用运行方式管理

任务全部运行

以下代码展示了通过run_all()忽略任务预设的时间安排，立即执行所有已定义的任务。先定义的任务先执行。示例如下：

import schedule
from datetime import datetime

def get_now_time():
    now = datetime.now()
    kEGHfBubnow = 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)

# 立即运行所有任务一次
schedule.run_all()

# 立即运行所有任务，每个任务运行间隔3秒
schedule.run_all(delay_seconds=3)

后台运行

默认情况下，无法在后台运行调度程序。不过，通过创建一个线程，利用该线程在不阻塞主线程的情况下运行任务。以下是实现这一操作的示例：

import threading
import time

import schedule

def run_continuously(interval=1):
    """
    创建一个后台线程持续运行调度器
    """
    # 创建事件对象用于控制线程停止
    # Event是事件类，线程间通信的简单机制，有"set"和"clear"两种状态
    # 初始状态为"clear"，通过cease_continuous_run.is_set()判断是否为set
    cease_continuous_run = threading.Event()

    class ScheduleThread(threading.Thread):
        @classmethod
        def run(cls):
            # 在线程启动后循环执行
            # 只要任务状态不是set，就运行任务
            while not cease_continuous_run.is_set():
                # 检查并执行所有待执行的定时任务
                schedule.run_pending()
                # 休眠指定间隔时间
                time.sleep(interval)

    # 创建并启动调度线程
    continuous_thread = ScheduleThread()
    continuous_thread.start()
    
    # 返回事件对象用于后续停止线程
    return cease_continuous_run

def background_job():
    """定时执行的后台任务"""
    print('Hello from the background thread')

# 设置定时任务：每秒执行一次background_job函数
schedule.every().second.do(background_job)

# 启动后台调度线程并获取停止控制器
stop_run_continuously = run_continuously()

# 主线程继续执行其他任务
print("主线程继续执行中...")
time.sleep(5)

# 停止后台调度线程
# 将事件对象的内部标志设置为set
stop_run_continuouslyphp.set()
print("后台线程已停止")

多任务同时执行

默认情况下，schedule任务调度工具会按顺序逐个执行所有任务。以10秒内执行30个任务为例，从日志中可以看到，这些任务会在这10秒内以串行方式依次执行，如同排队等候处理，而非同时运行。这种设计的核心目的是避免多个任务抢占资源或产生相互干扰，从而确保任务执行的稳定性和可靠性。

若需要实现多个任务并行运行，可通过为每个任务分配独立线程的方式达成，并通过统一队列进行调用。如下所示：

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

# 定义任务函数，打印当前时间和任务名称
def job(name):
    print(f"{get_now_time()} {name}")  

# 工作线程主函数，负责从队列中获取并执行任务
def worker_main():
    while True:
        joo_func,name = jobqueue.get()
        joo_func(name)
        jobqueue.task_done()

# 创建任务队列
jobqueue = queue.Queue()

# 调度多个相同间隔的任务，将任务放入队列
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"])

# 启动工作线程，对对任务队列进行处理
worker_thread = threading.Thread(target=worker_main)
worker_thread.start()

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

如果不需要队列统一调用，代码如下：

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

# 定义任务函数，打印当前时间和任务名称
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)

异常处理

调度程序不会捕获任务执行过程中发生的异常，并将异常传递给调用函数，可能直接崩溃程序：

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)

如果需要防范此类异常，可以按如下方式用装饰器封装任务函数：

import schedule  
import time     
import functools # 导入函数工具库

# 定义一个捕获异常的装饰器，用于包装定时任务
def catch_exceptions(cancel_on_failure=False):
    def catch_exceptions_decorator(job_func):
        @functools.wraps(job_func)  # 保留被装饰函数的元信息
        def wrapper(*args, **kwargs):
            try:
                return job_func(*args, **kwargs)  # 执行原函数
            except:
                import traceback
                print(traceback.format_exc())  # 打印完整的异常堆栈信息
                if cancel_on_failure:  # 如果设置了失败后取消任务
                    return schedule.CancelJob  # 返回取消任务的标志
        return wrapper
    return catch_exceptions_decorator

# 使用装饰器包装任务函数，设置失败后自动取消
@catch_exceptions(cancel_on_failure=True)
def bad_task():
    return 1 / 0 

schedule.every(1).minutes.do(bad_task)

# 主循环：持续检查并执行待处理的任务
while True:
    schedule.run_pending() 
    time.sleep(1)          

日志管理

通过logging库设置名为schedule的日志记录器并设置为DEBUG级别，使其能够捕获并处理schedule库内部产生的所有日志信息：

import schedule  
import logging  

# 配置基本日志设置
logging.basicConfig()
#python 获取名为'schedule'的日志记录器
# 由于schedule库在内部使用相同的名称（'schedule'）记录自己的日志
# 因此这个记录器可以捕获并处理schedule模块产生的所有日志信息
schedule_logger = logging.getLogger('schedule')
# 设置日志级别为DEBUG，以便记录详细的调试信息
schedule_logger.setLevel(level=logging.DEBUG)

def job():
    print("Hello, Logs")  # 打印信息到标准输出和log

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

# 立即运行所有已安排的任务（仅执行一次）
schedule.run_all()

# 清除所有已安排的任务
schedule.clear()

如果若想为任务添加可复用的日志记录功能，最简便的方法是实现一个处理日志的装饰器：

import functools
import time
import schedule

# 这个装饰器可用于任何任务函数，用于记录每次任务的执行时间
def print_elapsed_time(func):
    @functools.wraps(func) # 让被装饰函数的名称、文档字符串等属性保持不变。
    def wrapper(*args, **kwargs):
        # 记录任务开始时间戳
        start_timestamp = time.time()
        print(f'LOG: 正在运行任务 "{func.__name__}"')
        
        # 执行实际任务
        result = func(*args, **kwargs)
        
        # 计算并打印任务执行耗时
        print(f'LOG: 任务 "{func.__name__}" 已完成，耗时 {time.time() - start_timestamp:.1f} 秒')
        return rphpesult

    return wrapper

# 应用装饰器，自动记录该任务的执行时间
@print_elapsed_time
def job():
    print('Hello, Logs')
    # 模拟耗时操作
    time.sleep(2)

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

# 立即运行所有已注册的任务一次
schedule.run_all()

多调度程序运行

从一个调度程序里运行多少个任务都可以。不过要是调度程序规模比较大，可能需要用多个调度程序来管理。如下所示：

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")

# 主循环，使程序持续运行，不断检查并执行待处理的任务
while True:
    # 检查scheduler1中是否有待执行的任务，若有则执行
    scheduler1.run_pending()
    # 检查scheduler2中是否有待执行的任务，若有则执行
    scheduler2.run_pending()
    time.sleep(1)

以上就是Python中轻量级定时任务调度库schedule使用指南的详细内容，更多关于Python schedule库使用的资料请关注编程客栈(www.devze.com)其它相关文章！