### Python 定时任务调度执行脚本 #### 使用 `sched` 实现定时任务 `sched` 是 Python 内置的一个简单任务调度库，适合用于简单的场景。它通过时间戳来安排任务的执行顺序。 以下是使用 `sched` 调度一个定时任务的示例： ```python import sched import time def execute_script(message): print(f"{time.time()}: {message}") # 创建调度器实例 scheduler = sched.scheduler(time.time, time.sleep) # 添加任务到调度队列 scheduler.enter(5, 1, execute_script, argument=("Task executed after 5 seconds",)) scheduler.enter(10, 1, execute_script, argument=("Task executed after 10 seconds",)) # 启动调度器 scheduler.run() ``` 此代码将在指定的时间间隔后调用函数并打印消息[^2]。 --- #### 使用 `APScheduler` 实现定时任务 对于更复杂的需求，建议使用 `APScheduler`（Advanced Python Scheduler），这是一个功能强大且灵活的任务调度框架。它可以支持多种触发器模式（如日期、间隔和 Cron 表达式）以及不同的执行器（线程池或进程池）。 ##### 安装 APScheduler 可以通过 pip 命令安装 APScheduler： ```bash pip install apscheduler ``` ##### 示例：基于固定时间间隔的调度 以下是一个每隔 5 秒执行一次任务的例子： ```python from apscheduler.schedulers.blocking import BlockingScheduler def execute_script(): print("Executing the script...") # 创建调度器实例 scheduler = BlockingScheduler() # 添加任务，每5秒执行一次 scheduler.add_job(execute_script, 'interval', seconds=5) # 开始调度 try: scheduler.start() except (KeyboardInterrupt, SystemExit): pass ``` ##### 示例：基于 Cron 表达式的调度 如果需要按照特定的时间表（类似于 Linux 的 cron）执行任务，则可以使用 `CronTrigger`： ```python from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger def daily_task(): print("This task runs every day at 8 AM.") # 创建后台调度器实例 scheduler = BackgroundScheduler() # 设置每天早上8点执行任务 trigger = CronTrigger(hour='8', minute='0') scheduler.add_job(daily_task, trigger) # 开始调度 scheduler.start() print('Press Ctrl+C to exit') try: while True: time.sleep(2) except KeyboardInterrupt: scheduler.shutdown() ``` 以上代码会每天上午 8 点自动运行任务[^3]。 --- #### 配置执行器 为了提高性能，可以根据需求配置线程池或进程池作为执行器。例如： ```python from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor from apscheduler.schedulers.background import BackgroundScheduler executors = { 'default': ThreadPoolExecutor(20), # 最大允许20个线程并发 'processpool': ProcessPoolExecutor(3) # 最大允许3个进程并发 } scheduler = BackgroundScheduler(executors=executors) ``` 这样可以在高负载情况下更好地控制资源分配[^4]。 --- #### 异常处理与日志记录 在生产环境中，建议为调度器配置异常监听机制和详细的日志记录以便于调试问题。例如： ```python import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)-8s %(message)s', datefmt='%Y-%m-%d %H:%M:%S') def error_listener(event): if event.exception: logging.error(f'Job crashed with exception: {event.traceback}') else: logging.info('Job missed or rescheduled.') scheduler.add_listener(error_listener, EVENT_JOB_ERROR | EVENT_JOB_MISSED) ``` 这将捕获任何可能发生的错误，并将其写入日志文件中[^5]。 --- ### 总结 - 对于简单的定时任务，可以选择内置的 `sched` 库。 - 如果需要更多高级特性（如 Cron 支持或多线程/多进程执行），则推荐使用 `APScheduler`。 两者都可以满足不同层次的需求，具体选择取决于项目的复杂性和扩展性要求。