前阵子给手里的几个物联网传感节点写自动上报脚本,翻了下之前存的代码片段,发现光定时触发的实现我就用过好几种不同写法,索性整理下各自的踩坑点和实操步骤,省得每次选的时候来回翻文档。
最轻量化的循环休眠写法
不需要装任何第三方依赖,是所有写法里门槛最低的,核心逻辑就是写个死循环,执行完任务之后sleep固定时长。
import time
from datetime import datetime
def collect_sensor_data():
# 这里替换成你的实际业务逻辑,比如采集传感数据、上报接口
print(f"[{datetime.now()}] 数据上报完成")
if __name__ == "__main__":
while True:
collect_sensor_data()
# 间隔60秒执行,注意要把任务执行耗时算进去
time.sleep(60)
要注意这种写法的实际间隔是「任务执行时长+sleep时长」,如果任务本身跑的时间不稳定,定时误差会慢慢累计,而且多个任务并行的话要自己写多线程,维护起来很麻烦。
语法友好的第三方schedule库
这个库是专门给定时任务做的轻量封装,语法写起来很像自然语言,不用记复杂的参数,装完就能用。
pip install schedule
import schedule
import time
from datetime import datetime
def job1():
print(f"[{datetime.now()}] 每10秒执行的采集任务")
def job2():
print(f"[{datetime.now()}] 每天10点执行的日志清理任务")
schedule.every(10).seconds.do(job1)
schedule.every().day.at("10:00").do(job2)
while True:
schedule.run_pending()
time.sleep(1)
本质还是单线程阻塞运行,要是单个任务跑的时间太长,会影响后面的任务触发,而且默认没有异常捕获,某一个任务抛错的话整个定时循环都会直接退出,不适合无人值守的长期运行场景。
适合长期运行的APScheduler框架
如果是要跑在嵌入式网关、服务器上长期不重启的场景,这个库的功能会全很多,支持cron、间隔、固定时间点多种触发方式,自带异常处理、任务错过补偿、多线程执行,还可以把任务存在数据库里做持久化存储,重启之后不用重新加任务。
pip install apscheduler
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
def sensor_report():
print(f"[{datetime.now()}] 传感数据上报")
def log_clean():
print(f"[{datetime.now()}] 临时日志清理")
if __name__ == '__main__':
scheduler = BlockingScheduler(timezone='Asia/Shanghai')
# 间隔30秒执行
scheduler.add_job(sensor_report, 'interval', seconds=30)
# 每周一凌晨2点执行
scheduler.add_job(log_clean, 'cron', day_of_week='mon', hour=2)
scheduler.start()
具体选的时候其实不用纠结,对照自己的场景挑就行:
- 写临时跑的一次性脚本、demo验证的时候,直接用time循环就行,没有依赖问题,代码拷到哪都能跑,哪怕有几秒的定时误差也完全不影响。
- 个人写的小工具、本地跑的轻量任务,比如定时备份文件、爬个小网站数据,用schedule库足够,改触发时间的时候不用查复杂的配置语法,读代码的时候一眼就能看明白执行规则。
- 生产环境、长期运行的物联网网关、线上服务的定时任务,直接用APScheduler,不用自己补多线程、异常捕获、任务持久化的逻辑,省下来的时间摸鱼不好吗。
反正我现在写临时代码还是随手敲time.sleep,也没出过啥问题,工具嘛,适合当下的需求就好。
评论一下?