Python使用schedule库实现任务定时自动化

在Python的自动化工具库里，schedule可谓是个低调又强悍的存在。无论是定时备份数据、周期性抓取网页，还是定期发送提醒邮件，它都能让代码按计划精准执行，而不是靠人工盯着时钟。今天咱们就来掰开揉碎聊一聊schedule库的使用细节，帮你快速上手任务调度。

核心函数速览

schedule库的语法极简，却能把复杂的定时配置安排得明明白白。先通过pip安装：pip install schedule，然后来看它的核心函数与方法。

1. schedule.every() —— 创建基础调度任务

功能：开启一个新的调度任务，后续可以链式调用时间单位和执行函数。
案例：创建一个每分钟执行一次的任务框架。

import schedule
# 创建1个调度任务，默认参数为1
schedule.every()

2. 时间单位设置函数

功能：指定任务执行的时间间隔属性，包括seconds、minutes、hours、days、weeks，以及每周的某一天（monday、tuesday……sunday）。

案例：设置每5分钟执行一次的任务。

import schedule
# 每5分钟执行一次任务
schedule.every(5).minutes.do(task_function)
# 每2小时执行一次任务 
schedule.every(2).hours.do(task_function)
# 每3天执行一次任务
schedule.every(3).days.do(task_function)
# 每周执行一次任务
schedule.every().weeks.do(task_function)
# 每2周执行一次任务
schedule.every(2).weeks.do(task_function)
# 每周1执行一次任务
schedule.every().monday.do(task_function)
# 每周1和周3执行一次任务
schedule.every().monday.wednesday.do(task_function)
# 或者分开两个任务
schedule.every().monday.do(task_function)
schedule.every().wednesday.do(task_function)

3. at() —— 精准设置执行时间点

功能：在指定的时间点执行任务，常与days()、weeks()等结合使用。时间格式为HH:MM（24小时制），可以通过tz参数指定时区（需安装pytz）。

# 每天早上9点执行
schedule.every().day.at("09:00").do(task_function)
# 每周1早上9点执行
schedule.every().monday.at("09:00").do(task_function)
# 在东京时区，每周1早上9点执行
schedule.every().monday.at("09:00","Asia/Tokyo").do(task_function)

4. do() —— 指定要执行的函数

功能：关联具体的Python函数，当时间条件满足时自动触发执行。

案例：执行打印当前时间的函数。

import schedule
import time
def print_current_time():
    print(time.strftime("%Y-%m-%d %H:%M:%S"))
# 每10秒打印一次当前时间
schedule.every(10).seconds.do(print_current_time)

5. run_pending() —— 检查并执行待处理任务

功能：检查是否有到期任务，若有则执行。需要在循环中持续调用。

import schedule
import time
def task():
    print("任务执行！")
schedule.every(2).seconds.do(task)
while True:
    schedule.run_pending()

6. clear() —— 清除所有任务

功能：删除所有已设置的调度任务。

import schedule
schedule.every(1).minutes.do(task1)
schedule.every(2).minutes.do(task2)
# 清空所有任务
schedule.clear()

7. cancel_job(job) —— 取消特定任务

功能：移除指定的调度任务，job是do()方法的返回值。

import schedule
def task():
    print("任务执行！")
job = schedule.every(3).seconds.do(task)
# 取消任务
schedule.cancel_job(job)

案例实战

案例 1：定时发送邮件提醒

import schedule
import time
import smtplib
from email.mime.text import MIMEText

def send_email():
    msg = MIMEText('这是定时发送的邮件提醒！', 'plain', 'utf-8')
    msg['Subject'] = '定时邮件'
    msg['From'] = 'xxx@xxx.com'
    msg['To'] = 'xxx@xxx.com'
    try:
        server = smtplib.SMTP('smtp.qq.com', 587)
        server.starttls()
        server.login(msg['From'], 'xxxpassword')
        server.sendmail(msg['From'], msg['To'], msg.as_string())
        server.quit()
        print("邮件发送成功！")
    except Exception as e:
        print(f"邮件发送失败: {e}")

# 每天早上8点发送邮件
schedule.every().day.at("08:00").do(send_email)

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

案例 2：定时数据库备份

import schedule
import time
import subprocess

def backup_data():
    backup_command = "mysqldump -u username -ppassword database_name > backup.sql"
    try:
        subprocess.run(backup_command, shell=True, check=True)
        print("数据库备份成功！")
    except subprocess.CalledProcessError as e:
        print(f"数据库备份失败: {e}")

# 每隔2天执行一次数据备份任务
schedule.every(2).days.do(backup_data)

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

避坑指南

1. 循环阻塞问题

问题： schedule.run_pending()需要在循环中调用，但循环可能会阻塞主线程，导致其他代码无法运行。

解决：使用多线程或多进程运行调度循环，避免影响主程序执行。

import schedule
import time
import threading

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

# 启动调度循环线程，不影响主线程
schedule_thread = threading.Thread(target=run_schedule)
schedule_thread.start()

2. 时区与时间格式

问题：使用at()方法设置时间时，默认使用系统时区，可能与预期不符。

解决：如需跨时区调度，可结合pytz库进行时区转换，确保时间准确。以下是一些常用时区标识：

• 亚洲时区：Asia/Shanghai（北京时间）、Asia/Tokyo（东京）、Asia/Kolkata（加尔各答）
• 美洲时区：America/New_York（纽约）、America/Los_Angeles（洛杉矶）、America/Toronto（多伦多）
• 欧洲时区：Europe/London（伦敦）、Europe/Paris（巴黎）、Europe/Moscow（莫斯科）
• 大洋洲时区：Australia/Sydney（悉尼）、Pacific/Auckland（奥克兰）
• 非洲时区：Africa/Cairo（开罗）、Africa/Johannesburg（约翰内斯堡）

掌握schedule库的这些要点，就可以轻松实现各类定时任务自动化。无论是个人项目还是工作流程优化，它都能成为得力助手。