Python如何实现定时任务_多种方式实现定时任务

介绍


实现定时任务是一项基本的功能,Python中有很多的实现方案,以定时监测电脑的性能信息(如cpu、内存、磁盘等)为例,比较下这些方案的特点以及适合场景。

准备工作


使用pip下载 psutil 模块:

pip install psutil

实现监测性能的方法,代码如下:

import psutil
import datetime

def monitor_sys():
    # cpu、虚拟内存、disk等占比
    cpu_percent = psutil.cpu_percent()
    v_mem_percent = psutil.virtual_memory().percent
    c_disk_percent = psutil.disk_usage("C:\\").percent
    now_time_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"{now_time_str} cpu:{cpu_percent}, mem:{v_mem_percent}, disk:{c_disk_percent}")

定时任务实现方案

1、sleep + while


通过 time 模块的sleep()和while循环,实现定时执行任务,代码如下:

import time

def sleep2Schedule():
    while True:
        monitor_sys()
        # 5s调度一次
        time.sleep(5)
    # 递归调用    
    sleep2Schedule()

if __name__ == '__main__':
    sleep2Schedule()

输出结果:

2021-08-22 21:05:23 cpu:4.6, mem:75.9, disk:61.8
2021-08-22 21:05:29 cpu:5.8, mem:75.9, disk:61.8
2021-08-22 21:05:34 cpu:6.0, mem:75.8, disk:61.8
2021-08-22 21:05:39 cpu:4.4, mem:75.9, disk:61.8

优缺点:

优点是简单易实现,缺点也很明显,只能实现单一任务,不能异步执行。

2、线程模块的Timer()


通过 threading 模块的定时器Timer(),利用线程提供的start()启动多个线程,可以实现多个任务的定时执行,代码如下:

from threading import Timer

def monitor_sys():
    ......  # 省略
    print(f"{now_time_str} cpu:{cpu_percent}, mem:{v_mem_percent}, disk:{c_disk_percent}")
    Timer(5, monitor_sys).start()

def test_out():
    print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 这该死的颜值啊!")
    Timer(3, test_out).start()

if __name__ == '__main__':
    monitor_sys()
    test_out()

输出结果:

2021-08-22 21:22:39 cpu:0.0, mem:75.4, disk:61.8
2021-08-22 21:22:39 这该死的颜值啊!
2021-08-22 21:22:42 这该死的颜值啊!
2021-08-22 21:22:44 cpu:4.7, mem:75.5, disk:61.8
2021-08-22 21:22:45 这该死的颜值啊!
2021-08-22 21:22:48 这该死的颜值啊!
2021-08-22 21:22:49 cpu:5.3, mem:75.3, disk:61.8
2021-08-22 21:22:51 这该死的颜值啊!

优缺点:
优点是可以异步的执行多个调度任务,缺点是只能指定时间间隔(执行频率),不能指定执行时间点或时间段,不能动态添加定时任务等。

3、任务调度模块schedule


先下载该模块:

pip install schedule

通过schedule模块创建需要周期或定时调度执行的Job,通过 run_pending 方式执行任务,实现代码如下:

import schedule

def monitor_sys():
    ......  # 省略
    print(f"{now_time_str} cpu:{cpu_percent}, mem:{v_mem_percent}, disk:{c_disk_percent}")

def test_out():
    print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 这该死的颜值啊!")

def schedule2Schedule():
    schedule.clear()
    # 创建Job1,5s执行一次 monitor_sys
    schedule.every(5).seconds.do(monitor_sys)
    # 创建Job2,3s执行一次 test_out
    schedule.every(3).seconds.do(test_out)
    # 开启调度任务
    while True:
        schedule.run_pending()

if __name__ == '__main__':
    schedule2Schedule()

输出结果:

2021-08-22 21:38:49 这该死的颜值啊!
2021-08-22 21:38:51 cpu:15.5, mem:75.1, disk:61.8
2021-08-22 21:38:52 这该死的颜值啊!
2021-08-22 21:38:55 这该死的颜值啊!
2021-08-22 21:38:56 cpu:16.0, mem:75.2, disk:61.8
2021-08-22 21:38:58 这该死的颜值啊!
2021-08-22 21:39:01 cpu:18.1, mem:75.2, disk:61.8
2021-08-22 21:39:01 这该死的颜值啊!
2021-08-22 21:39:04 这该死的颜值啊!
2021-08-22 21:39:06 cpu:16.0, mem:75.1, disk:61.8

优缺点:

优点是可以异步执行多个定时任务,而且可以指定执行时间间隔(频率,如按分/按天等),时间点(如从8:00开始)及时间段(如5~10分钟等),缺点是因为轻量级,无法动态添加任务,无法将任务的结果进行持久化等。

4、任务框架APSchedule


APSchedule模块用于执行周期或定时任务,可以动态的添加或删除定时任务,还可以将任务的结果持久化到数据库、内存、队列等保存,此外,该还提供了丰富的接口去实现定时任务。

通过pip引入APSchedule模块:

pip install APScheduler

实现代码如下:

from apscheduler.schedulers.blocking import BlockingScheduler

def APSchedule2Schedule():
    # 创建调度器
    schedule = BlockingScheduler()
    # 动态添加任务: 触发器类型有``date``, ``interval`` or ``cron``,5s/次的话选择interval类型
    schedule.add_job(monitor_sys, trigger='interval', seconds=5)
    # 指定22:13执行的话选择interval类型
    schedule.add_job(test_out, trigger='cron', hour=22, minute=13)
    # 执行
    schedule.start()

if __name__ == '__main__':
    APSchedule2Schedule()

输出结果:(第二个任务按时间点,只执行了一次)

2021-08-22 22:12:55 cpu:4.7, mem:75.8, disk:61.8
2021-08-22 22:13:00 这该死的颜值啊!
2021-08-22 22:13:00 cpu:5.9, mem:75.6, disk:61.8
2021-08-22 22:13:05 cpu:3.8, mem:75.7, disk:61.8
2021-08-22 22:13:10 cpu:4.8, mem:75.6, disk:61.8
2021-08-22 22:13:15 cpu:5.1, mem:75.7, disk:61.8

5、Celery实现调度任务


最后,推荐一款可以比肩APSchedule的任务调度框架Celery,它具备APSchedule的优点,也支持动态增删Job,支持持久化等操作。Celery使用起来比APSchedule更复杂点,可以通过以下命令启动并监听定时任务:

python test_task.py celery worker --loglevel=info --beat

总结一下


经过上面实现定时任务的多种方式梳理,不难看出APSchedule框架是最优的选择,它简单易用,而且功能丰富。

版权声明:本文为作者原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。

原创文章,作者:老C,如若转载,请注明出处:https://www.code404.icu/1143.html

发表评论

登录后才能评论