ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Python定时任务实践:Schedule库的用法、坑位与工程化部署

Python定时任务实践:Schedule库的用法、坑位与工程化部署 1. 先把定时任务这件事想清楚Schedule 库能解决什么不适合什么1.1 为什么我在一堆轮子里挑中 Schedule如果你跟我一样手头有一堆“每天九点跑一下”“每半小时拉一次数据”的 Python 活你迟早会站在系统 cron 和一堆调度库面前做选择题。我最终用得最顺手的还是那个名字朴实无华的 Schedule 库。它不搞分布式、不引入 Redis、不写配置文件一条 pip 装完直接在代码里定义几行像话的时间规则整个定时任务就跑起来了。这篇文章不是给你念一遍官方文档而是我把它用在爬虫采集、量化行情拉取、后台报表生成等场景之后的完整复盘里面包含了传参、取消、手动触发、并发防重、异常和日志、服务器守护这些“把脚本变成服务”的细节。适合刚接触 Python 定时任务的新人也适合已经在写 Celery Beat 但只想轻量搞定单机需求的老手。先说我最看重的一点Schedule 的语法是“说人话”的。schedule.every(10).minutes.do(job)这种写法读一遍就知道是每 10 分钟跑一次 job不需要像 cron 表达式那样去背*/10 * * * *的段位含义。对一个经常要临时改任务的人来说这种可读性带来的维护成本下降是实打实的。尤其你现在接手一个三个月前写的脚本打开文件扫一眼每行 schedule 开头的代码就能立刻搞明白当时自己到底想在什么时间做什么事。还有一点很关键它是纯 Python 进程内调度不依赖系统服务。这意味着你在自己的开发机上写完脚本python xxx.py一跑就能调试不需要去改 /etc/crontab也不用担心改了系统配置影响同一台机器上的其他项目。对一个单机脚本、管理后台、采集任务来说这已经覆盖了 90% 的使用场景。1.2 先认清边界它到底不是一台调度服务器不过我必须把丑话说在前面。Schedule 这个名字容易让人产生错觉以为装完它就可以高枕无忧。实际它是一个“进程内调度器”不启动额外进程、不创建默认线程、不持久化任务也不会像操作系统服务那样维护一个后台守护。你的 Python 程序活着任务才可能触发程序一退出所有计划全部归零。所以真正使用它的前提是你的业务进程本来就打算长期运行或者你能用 systemd、nohup、Windows 计划任务等方式把脚本守护起来。它适合“脚本启动后自己往后排”的模式不适合跨机器、多实例、高可用、需要任务记录和重试的复杂调度。你要是把一个已经用 Schedule 写好的任务部署到两个 Web 进程里同一时刻每个进程都会执行一遍重复消息重复写入会立刻让你体会到什么叫“定时任务爆炸”。还有几个容易被忽略的点它没有内置的时区管理虽然高版本at()里可以带上 tz 参数但整体时区逻辑仍然偏简单它没有失败重试机制任务抛异常了就是抛异常后续靠你自己兜底它也没有分布式锁。理解这些边界以后你才能判断把 Schedule 用在哪些项目里是合适的哪些项目里它是扛不住的。1.3 一张表看清schedule 和 cron、APScheduler、Celery Beat 的分工方案运行形态优势短板schedulePython 进程内随脚本启停API 简单、零服务依赖、秒级到周级粒度无持久化、无分布式、需自建守护系统 cronOS 级后台调度系统稳定、不依赖具体语言语法别扭、调试不便、日志分散APScheduler进程内可配持久化存储触发器种类多支持 cron / date / interval线程模型需要理解配置稍重Celery Beat独立调度服务 broker支持分布式、任务队列天然隔离要引入 Redis 等中间件学习成本高看到这里你应该明白了Schedule 不是拿来替代谁的而是帮你在“简单但必须准时”这件事上快速交卷。我一般的原则是这样脚本和小服务用 Schedule系统级定时、需要跨机器协调就用 APScheduler 或 Celery纯本机又不介意看手册cron 当然也可以继续用。选型不丢人知道自己这个阶段需要什么才不丢人。2. 五分钟跑通安装与基础调度2.1 安装与环境检查别再栽在“python 未找到”上安装 Schedule 本身非常简单难的是你先把 Python 环境梳理明白。很多人照着教程第一行就是pip install schedule结果 Windows 终端里直接报一句“Python was not found; run without arguments to install from the Microsoft Store”然后就懵了。这句话的意思是系统里没有配置好 Python 解释器的 PATH或者压根没装 Python。我的建议是先确认解释器版本python --version # 或者 python3 --version如果命令不存在去 Python 官网下载安装包安装过程中一定要勾选“Add Python to PATH”选项。装完以后再确认 pip 可用然后安装库python -m pip install schedulepython -m pip这种写法其实比直接敲pip更稳因为你会发现把 Python 和 pip 绑定的是同一个解释器避免出现“pip 装到了 A 解释器代码却用 B 解释器跑”的经典问题。装完可以顺手验证一下python -c import schedule; print(schedule.__version__)能打印版本号说明安装成功。如果报ModuleNotFoundError: No module named schedule基本就是装错了解释器环境或者虚拟环境没有激活。这里我强烈建议项目用虚拟环境别把依赖一股脑装进全局环境。后面踩坑概率至少低一半。2.2 第一个调度器从“每 10 分钟跑一次”开始最简单的定时任务代码长这样import schedule import time def job(): print(定时任务执行, time.strftime(%Y-%m-%d %H:%M:%S)) schedule.every(10).minutes.do(job) while True: schedule.run_pending() time.sleep(1)这里最核心的是下面这个主循环while True: schedule.run_pending() time.sleep(1)Schedule 默认不会自己在后台“滴答”它只负责把到期的任务标记为 pending。你必须主动调用schedule.run_pending()它才会去看一眼有哪些任务该执行了然后把它们连同参数一起跑掉。time.sleep(1)是让主循环每秒钟检查一次既不让 CPU 空转又能保证定时触发的粒度在秒级。你可能会问为什么不直接把 sleep 去掉那这个 while 循环会以极快的频率空转CPU 占用直接飙升。对服务器来说这就是一种不必要的浪费。sleep(1) 是一个实践中最均衡的选择年收益是秒级精度成本几乎为零。跑起来以后你会看到程序每隔 10 分钟打一行时间。想停下来就用 CtrlC 中断。如果脚本是在后台跑的那就要用进程管理工具来收这个我在第四节和第五节都会再展开。2.3 时间表达人性化语法拆解Schedule 最舒服的地方就是时间表达。它提供的单位很全常用的基本覆盖了schedule.every(30).seconds.do(job) # 每 30 秒 schedule.every(2).hours.do(job) # 每 2 小时 schedule.every(3).days.do(job) # 每 3 天 schedule.every(4).weeks.do(job) # 每 4 周 schedule.every().minute.do(job) # 每分钟 schedule.every().hour.do(job) # 每小时 schedule.every().day.at(09:00).do(job) # 每天 9 点 schedule.every().day.at(09:00:30).do(job) # 每天 9 点 0 分 30 秒 schedule.every().monday.at(08:30).do(job) # 每周一 8:30 schedule.every().wednesday.at(13:15).do(job) # 每周三 13:15星期一到星期天分别对应monday到sunday写法非常直白。at()里的时间是 24 小时制别写09:00 PM这种格式。这里的.at(09:00:30)在需要秒级精确度的场景里很实用比如每周五 16:59:50 拉一次行情收盘快照。还有两个不太常用但很聪明的操作.to()和.until()。schedule.every(2).hours.to(4).hours.do(job)它的含义有点反直觉任务每隔 2 小时触发一次但触发的具体时间点会在 2 小时到 4 小时之间随机偏移。这个功能在分散负载时很实用多个任务同时启动想错峰用它就不用人工手工加偏移了。不过注意.to()只支持间隔型任务不能用在.at()那种固定时间点上。schedule.every(30).minutes.until(2030-01-01 00:00:00).do(job)until()可以给任务设置一个截止时间时间一到这个任务就不参与调度了。它接受字符串、datetime、timedelta等类型用来做“活动只持续到今天 23:59”这种临时任务非常顺手。3. 进阶用法传参、取消任务、按条件运行与重复任务3.1 给任务传参的正确姿势定时任务当然不只是print一下。现实里你往往要把函数参数传进去比如清理某个指定目录、给某个用户发通知。Schedule 的.do()支持直接传参import schedule def clean_dir(path, max_days7): print(f清理 {path} 中超过 {max_days} 天的文件) schedule.every().day.at(03:00).do(clean_dir, /var/tmp, 7)这里有个新手常犯的错写成了do(clean_dir(/var/tmp, 7))。你一旦在do()里加了括号函数就在注册那一刻立即执行了而不是被注册成定时任务。等调度器真正跑的时候它调用的反而是clean_dir的返回值如果你的函数没返回可调用对象任务就直接报错。正确姿势是只传函数对象参数放在.do()后面的位置参数或关键字参数里。如果任务需要的参数是动态变化的比如每次从数据库读最新配置那更推荐在任务函数内部去取而不是在注册时取一次然后硬编码。原因很简单注册那一刻读到的值和任务真正执行那一刻可能已经不一样了。你在调度器里留个“每次执行时再读取”的口子维护成本会低很多。3.2 任务注册与单独执行后台管理系统的“手动跑一次”“像 likeadmin 这种后台系统里加了一个定时任务但管理员想立刻手动执行一次怎么办”这个问题经常有人问。其实它的技术本质不是 Schedule 库的问题而是你的任务注册方式问题。定时调度要引用任务函数手动触发也要引用同一个任务函数。如果你把任务函数随手写在.do()里不去单独保存引用那手动触发时就只能翻代码重新找一遍。所以我建议做一个任务注册表import schedule import time task_registry {} def register_task(name): def decorator(func): task_registry[name] func return func return decorator register_task(report:send_daily) def send_daily_report(): print(发送每日报表) # 实际业务逻辑 register_task(cache:refresh) def refresh_cache(): print(刷新缓存) # 注册到调度器 schedule.every().day.at(09:00).do(send_daily_report) schedule.every(10).minutes.do(refresh_cache) # 手动执行后台管理系统里点按钮时调用 def manual_run(task_name): func task_registry.get(task_name) if func is None: raise ValueError(f任务不存在: {task_name}) print(f手动触发 {task_name}) return func()这样一来管理员在后台界面点“立即执行”按钮后端路由里调一下manual_run(report:send_daily)就行。定时调度和手动触发走的是同一个函数入口既不会出现两套逻辑也能保证手动执行的结果和定时执行完全一致。这个模式我建议直接沉淀成公共工具以后每个新任务只需要加一个register_task(任务名)装饰器调度和手动触发的入口自动就有了。如果你还想更精细一点可以在注册表里同时存任务描述、最近执行时间、执行状态。这样后端界面展示的“任务列表”和“执行历史”就都有数据来源了不用再靠打印日志去猜谁跑了谁没跑。3.3 取消任务、清理任务、设置截止时间任务不是注册了就铁定不能动。有时候你想临时停掉某个任务或者批量清空所有任务计划。Schedule 提供了几个操作job schedule.every(10).seconds.do(job_a) # 之后想取消 schedule.cancel_job(job) # 清空所有任务 schedule.clear() # 如果要清空某一类任务 schedule.clear(任务标签)这里重点提醒一下cancel_job()需要的是.do()返回的那个 Job 对象。如果你当时没有把它保存下来后面就要遍历schedule.jobs列表去筛选。所以从第一天起凡是你可能要在运行中取消的任务注册时一定要接住返回值cleanup_job schedule.every().day.at(02:00).do(clean_tmp)如果项目里任务很多我建议给 Job 对象自己加一个 tag。Schedule 的每个 Job 支持传入 tag方便你按业务域统一清理schedule.every().day.at(02:00).do(clean_tmp).tag(maintenance) schedule.clear(maintenance)注意.tag()要在.do()之后链式调用。这个机制在做“一键暂停所有后台任务”这种后台管理功能时特别好用不用一个接一个 cancel。3.4 装饰器式定义与复杂触发组合如果你更喜欢把任务定义和调度规则写在一起Schedule 也支持装饰器写法import schedule schedule.repeat(schedule.every().day.at(09:30)) def morning_job(): print(早上好开始干活)这段代码等价于def morning_job(): print(早上好开始干活) schedule.every().day.at(09:30).do(morning_job)装饰器在模块导入时就会把任务注册进 scheduler。有人喜欢这种写法觉得自律性强有人觉得它把“任务是什么”和“任务何时跑”耦合在一起了。我的看法是单人脚本无所谓随便你怎么写顺手但会在多文件项目里维护的调度任务还是推荐把调度规则集中在scheduler.py或者tasks.py里一眼能看完可维护性更好。复杂一点的组合需求比如“每个工作日的 9 点、12 点、18 点各跑一次”Schedule 没有直接表达“工作日”的内置概念但你可以叠加多个任务for h in [9, 12, 18]: schedule.every().monday.to().friday.at(f{h:02d}:00).do(job)这里的.monday.to().friday表达的是“周一 0 点到周五 23:59:59 这段时间窗口内的每小时档位”但它实际匹配逻辑需要你仔细测一遍。老实说我在这种“非标准时间窗口”上吃过亏如果你也要做工作日多时间点任务我更推荐直接写三个.monday()、.tuesday()……或者干脆用 APScheduler 的 cron 触发表达式。Schedule 的强项是简单场景简单场景里别硬上复杂语法。4. 真实落地阻塞、并发、异常与日志一个都别漏4.1 主循环模型为什么必须有个 while True 来驱动Schedule 的任务执行模型是同步单线程的。你写while True: schedule.run_pending(); time.sleep(1)这个循环里一旦某个 job 开始执行它要等 job 跑完才回到循环继续检查下一轮。对这个模型我从经验里总结出一个结论主循环的健壮性决定了整个定时任务的可靠性。我有一次是把一个跑批任务直接挂在这个主循环上这个任务平均耗时 3 分钟但我给它设置的周期是 2 分钟。结果跑起来以后任务还没跑完下一轮run_pending()又发现它已经到点了于是再次执行同一个任务。两个任务开始互相嵌套、共享同一个临时文件最后数据全乱了。后来我改成每次执行前先判断上次是否结束才算把这个问题按住。如果你任务本身很快比如几十毫秒执行完这个同步模型完全够用。但凡是任务可能超过调度间隔就必须引入并发或防重叠机制。这是 Schedule 使用中最容易被忽略的坎绝大多数“定时任务越跑越慢”的案例都能回溯到这里。4.2 任务卡住了怎么办并发线程与防重叠最简单的并发方案是把耗时任务丢到子线程里执行import threading import schedule import time def real_job(): time.sleep(120) print(耗时任务完成) def job_proxy(): threading.Thread(targetreal_job, daemonTrue).start() schedule.every(10).seconds.do(job_proxy) while True: schedule.run_pending() time.sleep(1)这里job_proxy是调度器真正调用的函数它只做一件事把real_job放进后台线程然后立即返回。这样调度器不会被长任务卡住还能继续执行其他任务。用daemonTrue的理由是如果主进程要退出子线程不至于成为僵尸。但线程方案引入了一个新问题任务重叠。假如 real_job 要跑 5 分钟而它每 10 秒就被触发一次线程会越来越密集最后机器被拖垮。所以并发必须伴随防重叠控制。我推荐一个轻量开关写法import threading job_lock threading.Lock() def heavy_job(): if not job_lock.acquire(blockingFalse): print(上一次还没跑完本次跳过) return try: # 真正的耗时逻辑 time.sleep(60) finally: job_lock.release()blockingFalse的意思是抢不到锁就算了而不是傻等。这样你就可以保证同一时刻这个任务只有一个实例在跑。如果你需要更复杂的并发策略比如允许最多 3 个实例并行跑那就上threading.Semaphore(3)原理一样只是信号量换成了计数器。4.3 异常别裸奔job 里必须自己处理 try/exceptSchedule 不会帮你吞掉异常。任务函数内部一旦抛出异常这个异常会一路冒泡到调用schedule.run_pending()的那个主循环如果你主循环没有 try/except整个进程就可能直接崩掉。对生产环境来说这等于“定时任务挂了什么日志都没留下”。所以我建议所有任务函数都自己包一层异常处理import logging logger logging.getLogger(scheduler.task) def job(): try: # 业务逻辑 pass except Exception: logger.exception(任务执行失败) # 这里可以决定是重试、补发告警还是静默跳过这个习惯看起来很笨但真的救过我很多次。里层只记录异常不让它往上冒主循环就一直是稳的。如果你的项目里任务数量很多还可以再进一步给异常处理加一个“连续失败 N 次后告警”的计数器。比如爬虫采集任务连续失败 3 次说明目标站点或数据库可能出了问题这时候直接发一条钉钉或企业微信通知比看日志快得多。有一点要特别注意如果你想在异常时重试重试逻辑一定要带重试次数上限和退避间隔否则网络一抖动任务会在短时间内疯狂重试把自己乃至依赖的下游系统打垮。我用过一个指数退避工具函数失败后按 1 次、2 次、4 次……的间隔递增重试封顶 8 次。你可以在自己的公共模块里也封装一个。4.4 日志系统接入让定时任务的运行可追踪日常脚本里你可能会直接 print 一堆东西但这些输出一旦到了后台要么没人看要么被系统日志冲走。定时任务必须要有一个像样的日志方案。我用 Python 标准库的 logging 就能解决绝大部分需求import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, handlers[ logging.FileHandler(scheduler.log, encodingutf-8), logging.StreamHandler() ] ) logger logging.getLogger(scheduler) def job(): logger.info(任务开始) try: # 业务逻辑 logger.info(任务成功) except Exception: logger.exception(任务失败)FileHandler负责落盘StreamHandler负责让控制台也看得到。生产环境建议再用logging.handlers.TimedRotatingFileHandler按天切分日志文件防止单个日志文件无限膨胀from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler( scheduler.log, whenmidnight, backupCount30, encodingutf-8 )backupCount30表示保留最近 30 天的日志。对管理后台来说这已经足够支撑“任务什么时候跑的、跑成功没有、失败原因是什么”的日常排查。如果你想把日志接入企业微信、钉钉或者自建告警平台也很好办只要在handlers里再加一个自定义 Handler 即可业务代码完全不用动。5. 工程化集成与部署从脚本到长期运行的守护任务5.1 和 Web 项目共存FastAPI/Flask 里的非阻塞注入很多人想把 Schedule 放进 FastAPI 或 Flask 项目里然后顺手在应用启动时写了一段while True结果 Web 服务直接起不来。原因很简单while True会阻塞事件循环或请求处理。正确做法是把调度循环放到独立的后台线程里。这里我提供一个封装好的函数基本可以到处复用import schedule import threading import time def run_continuously(scheduler, interval1): stop_event threading.Event() def _run(): while not stop_event.is_set(): scheduler.run_pending() time.sleep(interval) thread threading.Thread(target_run, daemonTrue) thread.start() return stop_event在 FastAPI 的启动事件里这样用from fastapi import FastAPI app FastAPI() stop_event None app.on_event(startup) def on_startup(): global stop_event schedule.every().day.at(09:00).do(send_daily_report) stop_event run_continuously(schedule) app.on_event(shutdown) def on_shutdown(): if stop_event: stop_event.set()返回的stop_event是给进程优雅退出用的。你调用stop_event.set()后台线程就会在下一次循环检查时退出不会残留孤儿线程。如果你用的是 Flask原理一模一样只是把启动钩子换成before_first_request或者应用工厂里的地方。核心思想就一句话不要让调度器占据主线程也不要把 Web 服务和调度器揉在一个事件循环里。5.2 部署到服务器后台任务与守护进程把脚本部署到 Linux 服务器上很多人第一反应是nohup python scheduler.py 。这能跑但它不健康没有自动重启、没有开机自启、日志管理全靠重定向。我更推荐用 systemd。写一个 service 文件比如/etc/systemd/system/python-scheduler.service[Unit] DescriptionPython Schedule Daemon Afternetwork.target [Service] Userappuser WorkingDirectory/opt/myapp ExecStart/opt/myapp/venv/bin/python /opt/myapp/scheduler.py Restartalways RestartSec5 [Install] WantedBymulti-user.target然后执行sudo systemctl daemon-reload sudo systemctl enable python-scheduler.service sudo systemctl start python-scheduler.serviceRestartalways的作用是一旦脚本异常退出或进程被杀systemd 会在 5 秒后自动把它拉起来。这种守护级别nohup 是给不了的。平时你想看它有没有活着最直接的方法是ps aux | grep scheduler.py journalctl -u python-scheduler.service -f这两条命令分别对应“进程查看与信号”和“日志系统”的诉求前一个确认进程状态后一个滚动看输出。如果你发现日志增长慢还想看内存和 CPU 情况再用top -p pid或者free -h快速判断是否出现泄漏。这些都是运维定时任务时最常用的一套组合拳。如果是 Windows 服务器那就用 Windows 计划任务在“任务计划程序”里创建基本任务操作里选“启动程序”程序填C:\Python312\python.exe参数填你的scheduler.py完整路径然后设定触发条件为“计算机启动时”或每天定时。核心差别只是没有 Linux 的 systemd 那么统一但思路是一样的。5.3 分布式场景怎么看为什么 Spring Cloud 那套方案和 Schedule 不是一回事有人会问那像 Java 生态里 Spring Cloud 架构中常见的分布式定时任务方案和这个 Schedule 库有什么关系答案很明确基本没关系它们解决的是两个完全不同维度的问题。Spring Cloud 体系的定时任务往往跑在微服务集群上同一套服务会有多个实例。如果每个实例都用本地调度同一个任务就会在每个实例上各执行一次造成重复操作。所以 Java 生态会引入 Quartz 集群、XXL-Job、Elastic-Job 这类东西用数据库或注册中心来协调“这一次该由哪个节点执行”。Python 生态里类似的选择是 Celery Beat 加 Redis或者 APScheduler 配合外部存储。Schedule 库本身没有任何跨节点协调能力它的任务列表只存在当前进程内存里。如果你的 Python 服务也是多实例部署我建议三条路把定时任务独立成一个单独的进程或单独的服务节点保证同一时刻只有一个实例在跑调度引入 Redis 分布式锁每个实例执行前先抢锁抢到了才执行迁移到 Celery Beat 或 APScheduler 共享存储。第 1 种最简单第 2 种适合任务不多、临时过渡的场景第 3 种适合认真做分布式。Schedule 库在这个选择里更合适的角色是第 1 种方案里的“进程内调度引擎”它帮我们把任务规则写清楚但分布式协调这件事它确实不做。6. 常见问题与排查实录我踩过的那些坑6.1 任务不执行先按这个思路排查遇到定时任务不执行我的排查顺序基本固定。先看主循环有没有跑起来。很多人把schedule.every().day.at(09:00).do(job)写在文件里文件跑完就结束了根本没有while True守着任务自然一次都不执行。这是一半以上“不执行”问题的原因。再看时间对不对。at()是 24 小时制09:00没问题9:00 PM就不行。还有时区问题如果服务器时区是 UTC你写的“09:00”其实是 UTC 的 9 点对应北京时间的下午 5 点。我吃过这个亏后来养成习惯先用date命令确认服务器时区再决定脚本里写本地时间还是 UTC。最后看进程是不是还活着。如果脚本跑在 systemd 下用journalctl -u python-scheduler.service看输出如果是 nohup 起的就去看对应日志文件。进程死了、日志也没写那就是被异常搞崩了回到第四节那个异常捕获的问题处理。6.2 安装与环境相关奇坑清单这节写给刚入门的读者。安装 Schedule 时最常见的报错和解决方法我整理成清单“Python was not found; run without arguments to install from the Microsoft Store”Windows 没配好 Python 的 PATH要么重新安装 Python 并勾选 Add to PATH要么手动把 Python 安装目录加进系统环境变量的 Path。“ModuleNotFoundError: No module named schedule”pip 装到了别的解释器或虚拟环境。用python -m pip install schedule确保当前解释器能装进当前环境。“pip 不是内部或外部命令”说明 pip 没在 PATH 里也可能 Python 根本没装成功。用python -m pip代替纯pip就能绕开。国内网络下 pip 下载慢临时加镜像源比如python -m pip install schedule -i https://pypi.tuna.tsinghua.edu.cn/simple。另外你如果问“Python 的库到底装在哪”可以先在解释器里查import sys print(sys.path)这里列出的目录就是当前解释器搜索模块的路径。如果代码能 import说明库就在其中某个目录下如果 import 失败多半是因为当前解释器和装库时用的解释器不是同一个。用虚拟环境能从根本上避免这类问题。6.3 时区、精度与文档陷阱Schedule 文档很精简但精简也意味着它把很多责任交给了你。时区是一个。at()在较新的版本里支持传 tz 参数比如schedule.every().day.at(09:00, Asia/Shanghai)建议跨时区部署时显式指定不要默认依赖服务器本地时区。精度是另一个。sleep(1)决定了调度检查的最小粒度因此你要的“每 30 秒触发”实际上会有几百毫秒到一两秒的漂移。对绝大多数定时任务来说这种漂移无伤大雅但如果你要跑高频交易撮合这类毫秒级任务Schedule 不是合适的工具建议直接上时间轮或专门的高精度调度器。还有几个隐藏的小坑.do()是注册动作不是立即执行所以千万别在参数里提前调用函数schedule.clear()会清掉所有任务如果你只想清一类任务先给任务加上 tag任务函数里如果开了数据库连接或文件句柄务必要在结束时关闭否则长期运行后连接数会涨到让人头大。最后分享一个我自己坚持的习惯无论 Schedule 脚本多简单我都会把日志、异常捕获、进程守护这三件事准备到位。爬虫项目半夜拉数据如果没有这三样内存一涨任务一崩第二天面对的就是一整片空数据反过来把这三样准备好Schedule 基本可以做到“扔在服务器上几个月不碰”。这个库体积不大但它能帮你扛住的事情比它表面上看起来要多得多。
返回列表