【Python--定时任务的四种方法】-Toy模板网

这篇具有很好参考价值的文章主要介绍了【Python--定时任务的四种方法】。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

前言

当每隔一段时间就要执行一段程序，或者往复循环执行某一个任务，这就需要使用定时任务来执行程序。应用很广泛，可以实现程序的自动化，而不需要我们手动的在规定时间内执行。如爬虫，如定时器等等。以下将介绍几种pyhton常用的定时任务方法。

while True：+sleep()

import datetime
import time


def task():
    now = datetime.datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts)


def func():
    while True:
        task()
        time.sleep(5)


func()

【Python--定时任务的四种方法】

优点

执行简单，代码容易理解

缺点

只能实现同步任务，控制困难，而且sleep是个阻塞函数。应用面也小，只能设置间隔，而不能指定具体的时间点。

threading.Timer定时器

timer最就是定时器，可以启动多个定时任务，且是异步执行，所以不存在等待顺序执行问题。timer常用的几种方法：

方法	说明
Timer(interval, function, args=None, kwargs=None)	创建定时器
cancel()	取消定时器
start()	使用线程方式执行
join(self, timeout=None)	等待线程执行结束

from datetime import datetime
from threading import Timer


def task():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts)


def func():
    task()
    # 定义一个定时器
    # 注意timer的语法
    # Timer(interval, function, args=None, kwargs=None)
    t = Timer(3, func)
    t.start()


func()

【Python--定时任务的四种方法】

多线程执行


from datetime import datetime
from threading import Timer
import threading
 
 
def task():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts)
 
 
def func():
    task()
    t = Timer(3, func)
    t.start()
 
 
if __name__ == '__main__':
    for i in range(3):
        thread = threading.Thread(None, func)
        thread.start()

【Python--定时任务的四种方法】

优点

可以实现异步任务，是非阻塞的。

缺点

但当运行次数过多时，会出现报错：Pyinstaller maximum recursion depth exceeded Error Resolution 达到最大递归深度，然后想到的是修改最大递归深度，但是运行到达到最大CPU时，python会直接销毁程序。

Timeloop库执行定时任务

Timeloop是一个库，可用于运行多周期任务。这是一个简单的库，使用decorator模式在线程中运行标记函数。

from datetime import datetime, timedelta
from timeloop import Timeloop
 
tl = Timeloop()
 
 
def task():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts + '333!')
 
 
def task2():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts + "555555!")
 
 
@tl.job(interval=timedelta(seconds=2))
def sample_job_every_2s():
    task()
 
 
@tl.job(interval=timedelta(seconds=5))
def sample_job_every_5s():
    task2()

调度模块schedule

schedule是一个第三方轻量级的任务调度模块，可以按照秒，分，小时，日期或者自定义事件执行时间。
如果想执行多个任务，也可以添加多个task。


import schedule
from datetime import datetime
 
 
def task():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts)
 
 
def task2():
    now = datetime.now()
    ts = now.strftime("%Y-%m-%d %H:%M:%S")
    print(ts + '666!')
 
 
def func():
    # 清空任务
    schedule.clear()
    # 创建一个按3秒间隔执行任务
    schedule.every(3).seconds.do(task)
    # 创建一个按2秒间隔执行任务
    schedule.every(2).seconds.do(task2)
    while True:
        schedule.run_pending()
 
 
func()