Python怎样实现定时任务？schedule模块

schedule

schedule

schedule.every()...do()

schedule

schedule.run_pending()

schedule

pip install schedule

import schedule
import time
import datetime

# 定义你的任务函数
def hello_job():
    """一个简单的任务，每隔几秒钟打印一次消息。"""
    print(f"Hello, world! It's {datetime.datetime.now().strftime('%H:%M:%S')}")

def data_cleanup_job(data_source):
    """一个模拟数据清理的任务，可以接受参数。"""
    print(f"Running data cleanup for {data_source} at {datetime.datetime.now().strftime('%H:%M:%S')}")
    # 实际的数据清理逻辑会在这里

def daily_report_job():
    """每天固定时间运行的任务。"""
    print(f"Generating daily report... {datetime.datetime.now().strftime('%H:%M:%S')}")

# 注册你的任务
# 每隔10秒运行一次 hello_job
schedule.every(10).seconds.do(hello_job)

# 每隔1分钟运行一次 hello_job
schedule.every(1).minute.do(hello_job)

# 每隔5分钟运行一次 data_cleanup_job，并传递参数
schedule.every(5).minutes.do(data_cleanup_job, "production_db")

# 每天的特定时间运行 daily_report_job
# 注意：这里的时间是字符串，格式为"HH:MM"
schedule.every().day.at("10:30").do(daily_report_job)

# 每周一的特定时间运行另一个任务
schedule.every().monday.at("09:00").do(
    lambda: print(f"Weekly Monday meeting reminder! {datetime.datetime.now().strftime('%H:%M:%S')}")
)

print("Scheduler started. Press Ctrl+C to stop.")

# 主循环：不断检查并运行待执行的任务
while True:
    schedule.run_pending()  # 运行所有到期的任务
    time.sleep(1)           # 暂停1秒，避免CPU空转，节省资源

schedule.run_pending()

time.sleep(1)

schedule

while True

schedule

schedule.every()...do()

schedule

schedule

schedule

schedule

try...except

import schedule
import time
import datetime
import logging # 使用Python内置的日志模块

# 配置日志，便于追踪问题
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def unreliable_job():
    """一个可能失败的任务。"""
    try:
        logging.info(f"Attempting to run unreliable_job at {datetime.datetime.now().strftime('%H:%M:%S')}")
        # 模拟一个随机的错误
        if datetime.datetime.now().second % 7 == 0: # 每7秒钟模拟一次错误
            raise ValueError("Simulated network connection error!")
        print("Unreliable job completed successfully.")
    except Exception as e:
        logging.error(f"Error in unreliable_job: {e}", exc_info=True) # exc_info=True会打印完整的堆栈信息
        # 在这里可以添加更多错误处理逻辑，比如发送邮件通知、记录到数据库等

schedule.every(5).seconds.do(unreliable_job)

print("Scheduler started with enhanced error handling. Press Ctrl+C to stop.")
while True:
    schedule.run_pending()
    time.sleep(1)

logging

print

schedule

schedule

schedule

import threading

_job_running = False

def long_running_job():
    global _job_running
    if _job_running:
        print("Long running job is already running, skipping this scheduled run.")
        return

    _job_running = True
    try:
        print(f"Starting long running job at {datetime.datetime.now().strftime('%H:%M:%S')}")
        time.sleep(15) # 模拟一个耗时15秒的任务
        print("Long running job finished.")
    finally:
        _job_running = False

schedule.every(10).seconds.do(long_running_job)
# ... (主循环不变)

schedule

schedule

cron