[python] 轻量级定时任务调度库schedule使用指北

奄蜊 · 2025-6-3 00:01:20

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).hoursjob2 = 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 pytzschedule.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_pendingimport timefrom 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 scheduleimport timefrom schedule import every, repeat, run_pendingfrom 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 scheduleimport 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 scheduleimport 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 scheduleimport timefrom 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 schedulefrom datetime import datetime, timedelta, timefrom 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 scheduleimport timedef 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 schedulefrom 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 threadingimport timeimport scheduledef 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_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)# 停止后台调度线程# 将事件对象的内部标志设置为setstop_run_continuously.set()print("后台线程已停止")

复制代码

多任务同时执行
默认情况下，schedule任务调度工具会按顺序逐个执行所有任务。以10秒内执行30个任务为例，从日志中可以看到，这些任务会在这10秒内以串行方式依次执行，如同排队等候处理，而非同时运行。这种设计的核心目的是避免多个任务抢占资源或产生相互干扰，从而确保任务执行的稳定性和可靠性。
若需要实现多个任务并行运行，可通过为每个任务分配独立线程的方式达成，并通过统一队列进行调用。如下所示：

import timeimport threadingimport scheduleimport queuefrom 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 timeimport threadingimport scheduleimport queuefrom 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 scheduleimport 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 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()# 获取名为'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 functoolsimport timeimport 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 result return wrapper# 应用装饰器，自动记录该任务的执行时间@print_elapsed_timedef job(): print('Hello, Logs') # 模拟耗时操作 time.sleep(2)schedule.every().second.do(job)# 立即运行所有已注册的任务一次schedule.run_all()

复制代码

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

import timeimport schedulefrom 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

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册