Python中实现定时任务详解

更新时间：2023年07月13日 08:47:39 作者：hd092336

这篇文章主要介绍了Python中实现定时任务详解的相关资料,需要的朋友可以参考下

固定时间间隔执行任务

import time
import logging

logging.basicConfig(
    level=logging.debug,
    format="%(asctime)s.%(msecs)d | %(threadName)s | %(levelname)s - %(message)s"
)


def task():
    logging.info("Task Start.")
    time.sleep(1)
    logging.info("Task Done.")

time.sleep

第一种办法是最简单又最暴力。
那就是在一个死循环中，使用线程睡眠函数 sleep()。

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

上述的方法有几个问题：

阻塞主进程，这个也好解决，可以放到线程中去执行
一次只有一个task，这个也有解决办法，可以多启几个线程

threading.Timer

既然第一种方法暴力，那么有没有比较优雅点的方法？
Python 标准库 threading 中有个 Timer 类。
它会新启动一个线程来执行定时任务，所以它是非阻塞函式。

原理：线程中预置一个finished的事件，通过finished.wait等待固定时间间隔。
超时则执行任务。如果需要取消任务，可以调用Timer.cancel来取消任务。

from threading import Timer

t = Timer(task, 5)
t.start()

优点：
不阻塞主进程，task在线程中执行

缺点：
一个Timer只能执行一次task就结束了。
如果想循环，需要改造一下task函数。

from threading import Timer

def repeat_task():
    t = Timer(5, repeat_task)
    # 开始任务的位置决定了是任务之间等待固定间隔时间
    # 还是每个任务的开始等待固定间隔时间
    t.start()
    task()

这样可以循环执行，但是仍然只能一个线程一个任务。

Q：如何跳出循环
A：加入一个标识符：

sleep可以用一个布尔值来控制
Timer可以用一个Event来控制

固定时间点执行任务

以上是用内置的方法中比较简单的实现方式。简单的功能可以实现。

前面执行的都是指定间隔时间的定时任务，那怎么执行指定时间点的任务呢？

上面的方法是可以做到的。有两种思路，

前面我们指定了间隔时间，那指定时间点，就先计算当前时间到指定时间点的间隔时间
不管间隔时间，而是记录任务时间点，然后实时去检查是否到达指定时间点

思路有了，但是对于固定时间点执行任务的场景以及后面更复杂的场景，自己实现可能就变得复杂。

下面再介绍几个进阶的定时任务的实现方式，可以适应更复杂的业务场景。

sched

第三种方式是使用标准库中sched模块。sched 是事件调度器，
它通过 scheduler 类来调度事件，从而达到定时执行任务的效果。

简单示例如下：

import sched

schedule = sched.scheduler()  # 初始化 sched 模块的 scheduler 类
schedule.enter(10, 1, task)  # 增加调度任务
schedule.run()  # 开始调度任务

scheduler 提供了两个添加调度任务的函数:

enter(delay, priority, action, argument=(), kwargs={})该函数可以延迟一定时间执行任务。delay 表示延迟多长时间执行任务，单位是秒。
priority为优先级，越小优先级越大。两个任务指定相同的延迟时间，优先级大的任务会向被执行。
action 即需要执行的函数，argument 和 kwargs 分别是函数的位置和关键字参数。
scheduler.enterabs(time, priority, action, argument=(), kwargs={})添加一项任务，但这个任务会在 time 这时刻执行。因此，time 是绝对时间。其他参数用法与 enter() 中的参数用法是一致。

优点：

执行时间间隔和时间点执行任务
可以添加不同的任务
任务可以设置优先级

缺点：
scheduler 中的每个调度任务只会工作一次，不会无限循环被调用。如果想重复执行同一任务，需要重复添加调度任务即可。

import sched
import threading
from functools import wraps

s = sched.scheduler()
STOP_FLAG = threading.Event()

def repeat(interval):
    def wrapper(func):
        @wraps(func)
        def inner():
            if not STOP_FLAG.is_set():
                s.enter(interval, 0, inner)
            func()
        return inner

    return wrapper


@repeat(5)
def new_task():
    return task()

new_task()
t = threading.Thread(target=s.run)
t.start()

实现原理

当然我们仅仅学会怎么用还是不够的，不能知其然而不知其所以然。
介绍了那么多方法，那么多库，但是底层的实现逻辑都是差不多的。
一个完整的定时任务系统，有三个部分：

任务队列(task queue)
根据执行时间和优先级进行排序
排序sorted(): 任务一多，整个队列重排，性能堪忧
heapq.pop()：堆排序，只获取最高优先级的任务，不重复排序
调度器(scheduler)
执行器(executor)
从前到后，实现的内容越来越多，控制的精细度也就越来越高，实现的功能也就越来越丰富。

import sched
import time
import threading
from functools import wraps

from task import task
from typing import Callable

STOP_FLAG = threading.Event()


def timeloop(func: Callable, interval: int):
    while True:
        func()
        time.sleep(interval)


def repeat_task(func: Callable, interval: int):
    # 新建任务的前后，决定了是在任务结束后等待时间，还是固定间隔
    if not STOP_FLAG.is_set():
        t = threading.Timer(interval, repeat_task, args=(func, interval))
        t.start()
    func()


def timer_schedule(func: Callable, interval: int):
    repeat_task(func, interval)


s = sched.scheduler()


def repeat(interval):
    def wrapper(func):
        @wraps(func)
        def inner():
            if not STOP_FLAG.is_set():
                s.enter(interval, 0, inner)
            func()
        return inner

    return wrapper


@repeat(5)
def new_task():
    return task()


def sched_schedule(func: Callable, interval: int):
    func()
    t = threading.Thread(target=s.run)
    t.start()
    return t


def main(schedule):
    schedule_thread = schedule(new_task, 5)
    while True:
        try:
            time.sleep(1)
        except KeyboardInterrupt:
            STOP_FLAG.set()
            schedule_thread.join()
            break


if __name__ == "__main__":
    # main(timeloop)
    # main(timer_schedule)
    main(sched_schedule)

到此这篇关于Python中实现定时任务详解的文章就介绍到这了,更多相关Python中实现定时任务内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

python编写爬虫小程序
我们可以通过python 来实现这样一个简单的爬虫功能，把我们想要的代码爬取到本地。下面就看看如何使用python来实现这样一个功能。
2015-05-05
TensorFlow Autodiff自动微分详解
这篇文章主要介绍了TensorFlow Autodiff自动微分详解，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-07-07
对Python 窗体(tkinter)文本编辑器(Text)详解
今天小编就为大家分享一篇对Python 窗体(tkinter)文本编辑器(Text)详解，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2018-10-10
关于CUDA out of memory的解决方案
这篇文章主要介绍了关于CUDA out of memory的解决方案，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-02-02
使用django实现一个代码发布系统
这篇文章主要介绍了使用django实现一个代码发布系统,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2019-07-07
Python实现分数序列求和
今天小编就为大家分享一篇Python实现分数序列求和，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-02-02
Python 中设置请求的最大重试次数示例代码
本篇文章介绍了为什么我们会收到错误消息，指出超出了最大重试次数，以及我们如何在 Python 中为请求设置 max_retries，本文通过示例代码给大家介绍的非常详细，需要的朋友参考下吧
2023-06-06
django使用定时任务django_apscheduler的实现
定时任务无论是个人开发还是企业业务都是需要的，本文主要介绍了django使用定时任务django_apscheduler的实现，减少请求时需要用户等待的时间，感兴趣的可以了解一下
2021-08-08
python中取绝对值简单方法总结
在本篇内容里小编给大家整理的是关于python中取绝对值简单方法，需要的朋友们可以学习下。
2020-07-07
Django框架中方法的访问和查找
这篇文章主要介绍了Django框架中方法的访问和查找,包括各种列表和字典的查找,是Python的Django框架使用的必备技巧,需要的朋友可以参考下
2015-07-07