layout: posttitle: python定时任务
subtitle: 定时任务的四种实现方式 APScheduler
date: 2020-04-02
author: NSX
header-img: img/post-bg-ios9-web.jpg
catalog: true
tags:
- 技术
- 定时任务
- Python

Python定时任务

实现系统监测功能为例:

1：定时或者定点监测CPU与内存使用率；

2：将时间，CPU，内存使用情况保存到日志文件；

主要介绍4类开启定时任务的方法：

  • 最简单使用方式：循环+sleep

  • 线程模块中Timer类

  • 调度模块：schedule

  • 💡定时任务框架：APScheduler

导入需要的包

  1. # psutil:获取系统信息模块，可以获取CPU，内存，磁盘等的使用情况
  2. import psutil
  3. import time
  4. import datetime
  5. import schedule
  6. import time
  7. from threading import Timer
  8. from apscheduler.schedulers.blocking import BlockingScheduler

1. 最简单使用方式：循环+sleep

使用while+sleep就可以实现

  1. # logfile：监测信息写入文件
  2. def MonitorSystem(logfile=None):
  3.     # 获取cpu使用情况
  4.     cpuper = psutil.cpu_percent()
  5.     # 获取内存使用情况：系统内存大小，使用内存，有效内存，内存使用率
  6.     mem = psutil.virtual_memory()
  7.     # 内存使用率
  8.     memper = mem.percent
  9.     # 获取当前时间
  10.     now = datetime.datetime.now()
  11.     ts = now.strftime("%Y-%m-%d %H:%M:%S")
  12.     line = f"{ts} cpu:{cpuper}%, mem:{memper}%"
  13.     print(line)
  14.     if logfile:
  15.         logfile.write(line)
  18. def loopMonitor():
  19.     while True:
  20.         MonitorSystem()
  21.         # 3s检查一次
  22.         time.sleep(3)
  23. loopMonitor()

2. 线程模块中Timer类

timer最基本理解就是定时器，我们可以启动多个定时任务，这些定时器任务是异步执行，所以不存在等待顺序执行问题。

  1. # logfile：监测信息写入文件
  2. def MonitorSystem2(logfile=None):
  3.     # 获取cpu使用情况
  4.     cpuper = psutil.cpu_percent()
  5.     # 获取内存使用情况：系统内存大小，使用内存，有效内存，内存使用率
  6.     mem = psutil.virtual_memory()
  7.     # 内存使用率
  8.     memper = mem.percent
  9.     # 获取当前时间
  10.     now = datetime.datetime.now()
  11.     ts = now.strftime("%Y-%m-%d %H:%M:%S")
  12.     line = f"{ts} cpu:{cpuper}%, mem:{memper}%"
  13.     print(line)
  14.     if logfile:
  15.         logfile.write(line)
  16.     # 启动定时器任务，每三秒执行一次
  17.     Timer(3, MonitorSystem2).start()
  20. MonitorSystem2()

3. 调度模块：schedule

  • schedule是一个第三方轻量级的任务调度模块，可以按照秒，分，小时，日期或者自定义事件执行时间；

  • 安装方式：pip install schedule

  • 特征

    1. 一种易于使用的API，用于调度作业。

    2. 非常轻巧，没有外部依赖性。

    3. 出色的测试覆盖率。

    4. 在Python 2.7、3.5和3.6上测试

  1. def tasklist():
  2.     # 清空任务
  3.     schedule.clear()
  4.     # 创建一个按秒间隔执行任务
  5.     schedule.every(1).seconds.do(MonitorSystem)
  6.     # schedule.every(1).hour.do(MonitorSystem)
  7.     # 执行10S
  8.     while True:
  9.         schedule.run_pending()
  10.         time.sleep(1)
  13. tasklist()

4. 定时任务框架：APScheduler💡

官网文档：http://apscheduler.readthedoc…

API：http://apscheduler.readthedoc…

介绍：https://www.cnblogs.com/Neeo/p/10435059.html

APScheduler简介

APScheduler（Advanced Python Scheduler）是一个轻量级的Python定时任务调度框架（Python库）。APScheduler有三个内置的调度系统，其中包括：

  • cron式调度（可选开始/结束时间）
  • 基于间隔的执行（以偶数间隔运行作业，也可以选择开始/结束时间）
  • 一次性延迟执行任务（在指定的日期/时间内运行作业一次）

APScheduler组件

APScheduler 包含四个组件，分别是：

  • 触发器（trigger），触发器中包含调度逻辑，每个作业都有自己的触发器来决定下次运行时间。除了它们自己初始配置以外，触发器完全是无状态的。
  • 作业存储器（job store），存储被调度的作业，默认的作业存储器只是简单地把作业保存在内存中，其他的作业存储器则是将作业保存在数据库中，当作业被保存在一个持久化的作业存储器中的时候，该作业的数据会被序列化，并在加载时被反序列化，需要说明的是，作业存储器不能共享调度器。
  • 执行器（executor），处理作业的运行，通常通过在作业中提交指定的可调用对象到一个线程或者进程池来进行，当作业完成时，执行器会将通知调度器。
  • 调度器（scheduler），配置作业存储器和执行器可以在调度器中完成。例如添加、修改、移除作业，根据不同的应用场景，可以选择不同的调度器，可选的将在下一小节展示。

各组件简介

调度器

  • BlockingScheduler:阻塞调度。当你的程序只运行这个调度器时可以使用。
  • BackgroundScheduler:后台调度。你的应用程序不止运行调度器，且想它在应用程序中后台运行时，可以使用该调度器。
  • AsyncIOScheduler:如果 你应用的程序使用了asyncio模块，需要使用此调度器。
  • GeventScheduler:如果你的应用程序使用了gevent技术，需要使用此调度器。
  • TornadoScheduler:可以用于开发Tornado应用程序。
  • TwistedScheduler:可以用于开发Twisted应用程序。
  • QtScheduler:可以用于开发Qt应用程序。

作业存储器

如果你的应用在每次启动的时候都会重新创建作业，那么使用默认的作业存储器（MemoryJobStore）即可，但是如果你需要在调度器重启或者应用程序奔溃的情况下任然保留作业，你应该根据你的应用环境来选择具体的作业存储器。例如：使用Mongo或者SQLAlchemy JobStore （用于支持大多数RDBMS）

执行器

对执行器的选择取决于你使用上面哪些框架，大多数情况下，使用默认的ThreadPoolExecutor已经能够满足需求。如果你的应用涉及到CPU密集型操作，你可以考虑使用ProcessPoolExecutor来使用更多的CPU核心。你也可以同时使用两者，将ProcessPoolExecutor作为第二执行器。

触发器

当你调度作业的时候，你需要为这个作业选择一个触发器，用来描述这个作业何时被触发，APScheduler有三种内置的触发器类型：

  • date 一次性指定日期
  • interval 在某个时间范围内间隔多长时间执行一次
  • cron 和Linux crontab格式兼容，最为强大

APScheduler安装

pip 安装

  1. $ pip install apscheduler

源码安装

  1. $ python setup.py install

使用

当你需要调度作业的时候，你需要为这个作业选择一个触发器，用来描述该作业将在何时被触发，APScheduler有3中内置的触发器类型：

  • 新建一个调度器（scheduler）
  • 添加一个调度任务（job store)
  • 运行调度任务

添加作业#

有两种方式可以添加一个新的作业：

  • add_job来添加作业
  • 装饰器模式添加作业

只执行一次#

  1. import datetime
  2. from apscheduler.schedulers.blocking import BlockingScheduler
  3. def job2(text):
  4.     print('job2', datetime.datetime.now(), text)
  5. scheduler = BlockingScheduler()
  6. scheduler.add_job(job2, 'date', run_date=datetime.datetime(2019, 2, 25, 19, 5, 6), args=['text'], id='job2')
  7. scheduler.start()

上例中，只在2019-2-25 19:05:06执行一次，args传递一个text参数。

间隔执行#

下面来个简单的例子，作业每个5秒执行一次：

  1. import datetime
  2. from apscheduler.schedulers.blocking import BlockingScheduler
  4. def job1():
  5.     print('job1', datetime.datetime.now())
  6. scheduler = BlockingScheduler()
  7. scheduler.add_job(job1, 'interval', seconds=5, id='job1')  # 每隔5秒执行一次
  8. scheduler.start()

每天凌晨1点30分50秒执行一次

  1. from apscheduler.schedulers.blocking import BlockingScheduler  # 后台运行
  2. sc = BlockingScheduler()
  3. f = open('t1.text', 'a', encoding='utf8')
  4. @sc.scheduled_job('cron', day_of_week='*', hour=1, minute='30', second='50')
  5. def check_db():
  6.     print(111111111111)
  7. if __name__ == '__main__':
  8.     try:
  9.         sc.start()
  10.         f.write('定时任务成功执行')
  11.     except Exception as e:
  12.         sc.shutdown()
  13.         f.write('定时任务执行失败')
  14.     finally:
  15.         f.close()

每几分钟执行一次：

  1. import datetime
  2. from apscheduler.schedulers.blocking import BlockingScheduler
  4. def job1():
  5.     print('job1', datetime.datetime.now())
  6. scheduler = BlockingScheduler()
  7. # 每隔2分钟执行一次， */1：每隔1分钟执行一次
  8. scheduler.add_job(job1, 'cron', minute="*/2", id='job1') 
  9. scheduler.start()

每小时执行一次：

  1. import datetime
  2. from apscheduler.schedulers.blocking import BlockingScheduler
  4. def job1():
  5.     print('job1', datetime.datetime.now())
  6. scheduler = BlockingScheduler()
  7. # 每小时执行一次
  8. scheduler.add_job(job1, 'interval', hours=1, id='job1')
  9. # 每小时执行一次，上下浮动120秒区间内
  10. # scheduler.add_job(job1, 'interval', hours=1, id='job1', jitter=120)
  11. scheduler.start()

最后选择

简单总结上面四种定时定点任务实现：

  1. 循环+sleep方式适合简答测试，

  2. timer可以实现定时任务，但是对定点任务来说，需要检查当前时间点；

  3. schedule可以定点定时执行，但是需要在循环中检测任务，而且存在阻塞；

  4. APScheduler框架更加强大，可以直接在里面添加定点与定时任务；

综合考虑，老猫决定使用APScheduler框架，实现简单，只需要直接创建任务，并将添加到调度器中即可。

附录-Crontab命令

在CentOS系统上，通过 crontab 命令，安排那些需要被周期性执行的命令

crontab命令的语法为：

  1. crontab [-e [UserName]|-l [UserName]|-r [UserName]|-v [UserName]|File ]

各个参数说明：

  1. -e [UserName]: 执行文字编辑器来设定时程表，内定的文字编辑器是 vi
  2. -r [UserName]: 删除目前的时程表
  3. -l [UserName]: 列出目前的时程表
  4. -v [UserName]:列出用户cron作业的状态

crontab 文件格式

  1. {minute} {hour} {day-of-month} {month} {day-of-week} {full-path-to-shell-script}
  • minute: 分钟 (0-59)
  • hour: 小时 (0-23)
  • day-of-month: 日期 (0-31)
  • month: 月份 (0-12 [12 代表 December])
  • Day-of-week: 一周当中的某天 (0-7 [7 或 0 代表星期天])
  • full-path-to-shell-script: 计划执行的脚本或命令的名称

crontab命令中的一些常用特殊符号：

符号 说明
* 表示任何时刻
, 表示分割
－ 表示一个段，如第二段里： 1-5，就表示1到5点
/n 表示每个n的单位执行一次，如第二段里，*/1, 就表示每隔1个小时执行一次命令。也可以写成1-23/1.

Crontab 示例

每隔 5 分钟运行一次 backupscript 脚本

  1. */5 * * * * /root/backupscript.sh

每天的凌晨 1 点运行 backupscript 脚本

  1. 0 1 * * * /root/backupscript.sh

每月的第一个凌晨 3:15 运行 backupscript 脚本

  1. 15 3 1 * * /root/backupscript.sh

每星期六的晚上11 : 00 pm重启smb

  1. 0 23 * * 6 /etc/init.d/smb restart

特别注意：python的执行命令 必须用绝对路径！！

参考

https://blog.51cto.com/huangyg/2367088

Python任务调度模块APScheduler

Linux学习之CentOS(十二)—crontab命令的使用方法