[python] 轻量级定时任务调度库schedule使用指北
schedule是一款专为简化定时任务调度而设计的Python库，它通过直观的语法降低了周期性任务的实现门槛。作为进程内调度器，它无需额外守护进程，轻量且无外部依赖，适合快速搭建自动化任务。不过，该库在功能完整性上有所取舍，目前暂不支持断点续传、亚秒级精度控制以及多任务并行执行等复杂场景。
schedule库的官方仓库地址见：schedule，schedule库的官方文档见：schedule-doc。
schedule库支持在Python 3.7及以上版本的环境中运行，schedule库的安装命令如下：
pip install schedule
文章目录
- 1 使用入门
- 1.1 基础使用
- 1.1.1 相对调用
- 1.1.2 绝对调用
- 1.2 进阶使用
- 1.2.1 调用程序管理
- 1.2.2 调用时间管理
- 1.2.3 调用运行方式管理
- 2 参考
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 datetimedef get_now_time():now = datetime.now()now = now.strftime("%Y-%m-%d %H:%M:%S")return nowdef 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 datetimedef get_now_time():now = datetime.now()now = now.strftime("%Y-%m-%d %H:%M:%S")return nowdef 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 datetimedef 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 datetimedef get_now_time():now = datetime.now()now = now.strftime("%Y-%m-%d %H:%M:%S")return nowdef 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 scheduledef 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 timedef job_that_executes_once(name):print(f"hello {name}")return schedule.CancelJobschedule.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 timedef 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 datetimedef get_now_time():now = datetime.now()now = now.strftime("%Y-%m-%d %H:%M:%S")return nowdef 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 datetimedef get_now_time():now = datetime.now()now = now.strftime("%Y-%m-%d %H:%M:%S")return nowdef 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秒执行一次任务
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 timedef job():print('你好')schedule.every(2).minutes.do(job)while 1:# n为距离下次执行任务的剩余秒数n = schedule.idle_seconds()print(n)if n is None:# 没有更多任务breakelif n > 0:# 精确睡眠相应的时间time.sleep(n)schedule.run_pending()
1.2.3 调用运行方式管理
任务全部运行
以下代码展示了通过run_all()忽略任务预设的时间安排，立即执行所有已定义的任务。先定义的任务先执行。示例如下：
import schedule
from datetime import datetimedef get_now_time():now = datetime.now()now = now.strftime("%Y-%m-%d %H:%M:%S")return nowdef 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 timeimport scheduledef run_continuously(interval=1):"""创建一个后台线程持续运行调度器"""# 创建事件对象用于控制线程停止# Event是事件类，线程间通信的简单机制，有"set"和"clear"两种状态# 初始状态为"clear"，通过cease_continuous_run.is_set()判断是否为setcease_continuous_run = threading.Event()class ScheduleThread(threading.Thread):@classmethoddef 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_rundef 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_continuously.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 timedef bad_task():return 1 / 0schedule.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 tracebackprint(traceback.format_exc()) # 打印完整的异常堆栈信息if cancel_on_failure: # 如果设置了失败后取消任务return schedule.CancelJob # 返回取消任务的标志return wrapperreturn 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()
# 获取名为'schedule'的日志记录器
# 由于schedule库在内部使用相同的名称（'schedule'）记录自己的日志
# 因此这个记录器可以捕获并处理schedule模块产生的所有日志信息
schedule_logger = logging.getLogger('schedule')
# 设置日志级别为DEBUG，以便记录详细的调试信息
schedule_logger.setLevel(level=logging.DEBUG)def job():print("Hello, Logs") # 打印信息到标准输出和logschedule.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 resultreturn 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 nowdef 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)
2 参考
- schedule
- schedule-doc