Python使用schedule库实现定时运行py文件

pip install schedule

import sys
import time
from datetime import datetime

def run_task():
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{current_time}] [Worker] 开始执行日常统计同步任务...")
    
    # 模拟耗时操作，例如网络请求或数据库写入
    time.sleep(2)
    
    print(f"[{current_time}] [Worker] 任务完成，正常退出。")

if __name__ == "__main__":
    run_task()
    sys.exit(0)

python3 worker.py

import sys
import time
import schedule
import subprocess
from pathlib import Path

# 定位当前脚本所在目录及目标 py 文件的绝对路径
BASE_DIR = Path(__file__).resolve().parent
TARGET_SCRIPT = BASE_DIR / "worker.py"

def execute_py_file():
    '''触发并执行外部 py 脚本的回调任务'''
    if not TARGET_SCRIPT.exists():
        print(f"❌ 错误：找不到目标文件 {TARGET_SCRIPT}")
        return

    print(f"⏰ [Scheduler] 到达预定时间，准备启动子进程：{TARGET_SCRIPT.name}")

    try:
        # sys.executable 确保调用与调度器完全相同的 Python 解释器环境
        result = subprocess.run(
            [sys.executable, str(TARGET_SCRIPT)],
            capture_output=True,
            text=True,
            timeout=60,  # 设定最大执行限时，防止外部任务死锁
            cwd=str(BASE_DIR)  # 显式对齐工作目录，避免文件相对路径找不到
        )

        # 打印子进程的标准输出与标准错误
        if result.stdout:
            print("--- [子进程输出] ---")
            print(result.stdout.strip())
        
        if result.returncode == 0:
            print(f"✅ [Scheduler] 子进程执行成功 (退出码: {result.returncode})")
        else:
            print(f"⚠️ [Scheduler] 子进程返回异常退出码: {result.returncode}")
            if result.stderr:
                print(f"--- [错误详情] ---\n{result.stderr.strip()}")

    except subprocess.TimeoutExpired:
        print(f"⛔ [Scheduler] 任务执行超时（超过 60 秒），已强制终止")
    except Exception as e:
        print(f"❌ [Scheduler] 调用子进程发生未预料异常: {e}")

# 注册定时调度规则
# 示例 1: 每隔 10 秒执行一次（便于调试观察）
schedule.every(10).seconds.do(execute_py_file)

# 示例 2: 常用常规周期设定语法
# schedule.every(30).minutes.do(execute_py_file)
# schedule.every().hour.at(":15").do(execute_py_file)
# schedule.every().day.at("02:30").do(execute_py_file)
# schedule.every().monday.at("09:00").do(execute_py_file)

print("🚀 [Scheduler] 调度服务已启动，按 Ctrl + C 可终止...")

# 维持主轮询常驻循环
while True:
    schedule.run_pending()
    time.sleep(1)  # 适度休眠 1 秒，将 CPU 占用率降至近 0%

import sys
import time
import schedule
import threading
import subprocess
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
TARGET_SCRIPT = BASE_DIR / "worker.py"

# 防重复并发锁：防止上一次长任务还没跑完，下一个同名任务就无节制启动
task_lock = threading.Lock()

def _worker_process_runner():
    '''在后台线程中实际运行子进程的封装函数'''
    # 尝试获取锁；如果上一次还在跑，则果断跳过本次，避免服务器负载雪崩
    if not task_lock.acquire(blocking=False):
        print(f"⏳ [Scheduler] 上一轮任务尚未结束，跳过本次触发，防止并发堆叠。")
        return

    try:
        start_ts = time.time()
        print(f"▶️ [Worker Thread] 启动后台子任务: {TARGET_SCRIPT.name}")
        
        proc = subprocess.run(
            [sys.executable, str(TARGET_SCRIPT)],
            capture_output=True,
            text=True,
            timeout=300
        )
        
        duration = round(time.time() - start_ts, 2)
        if proc.returncode == 0:
            print(f"✔ [Worker Thread] 任务完成，耗时 {duration} 秒")
        else:
            print(f"❌ [Worker Thread] 任务失败 (退出码: {proc.returncode})\n{proc.stderr.strip()}")
            
    except subprocess.TimeoutExpired:
        print("⛔ [Worker Thread] 任务执行超时被强行终止！")
    except Exception as ex:
        print(f"❌ [Worker Thread] 运行异常: {ex}")
    finally:
        task_lock.release()

def run_async_job():
    '''调度器统一入口：每次触发直接生成后台子线程'''
    thread = threading.Thread(target=_worker_process_runner, daemon=True)
    thread.start()

# 设定每 10 秒运行一次，即便 worker 耗时 15 秒，也不会卡死主循环
schedule.every(10).seconds.do(run_async_job)

print("🚀 [Scheduler] 多线程非阻塞调度器运行中...")
while True:
    schedule.run_pending()
    time.sleep(1)

开启异步多线程后，务必评估是否允许同一任务多实例并发运行。若目标脚本涉及写同一份数据库、修改相同本地文件或调用限流 API，多个实例同时跑会发生数据覆盖或冲突。代码中通过 task_lock.acquire(blocking=False) 实现“前序未完则自动丢弃本期”，是工程实践中最简单有效的防雪崩机制。

[Unit]
Description=Python Schedule Task Runner
After=network.target

[Service]
Type=simple
User=www-data
WorkingDirectory=/var/www/my_project
ExecStart=/var/www/my_project/venv/bin/python scheduler_threaded.py
Restart=always
RestartSec=10
StandardOutput=append:/var/log/my_project/scheduler_stdout.log
StandardError=append:/var/log/my_project/scheduler_stderr.log

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now py-scheduler
sudo systemctl status py-scheduler