在項(xiàng)目中，我們可能遇到有定時(shí)任務(wù)的需求。

• 其一：每隔一個(gè)時(shí)間段就執(zhí)行任務(wù)。
  比如：壓測中每隔45分鐘調(diào)整溫箱的溫度。
• 其二：定時(shí)執(zhí)行任務(wù)。
  例如每天早上 8 點(diǎn)定時(shí)推送早報(bào)。

今天，我跟大家分享下 Python 定時(shí)任務(wù)的實(shí)現(xiàn)方法。

固定時(shí)間間隔執(zhí)行任務(wù)

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

第一種辦法是最簡單又最暴力。
那就是在一個(gè)死循環(huán)中，使用線程睡眠函數(shù) sleep()。

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

上述的方法有幾個(gè)問題：

• 阻塞主進(jìn)程，這個(gè)也好解決，可以放到線程中去執(zhí)行
• 一次只有一個(gè)task，這個(gè)也有解決辦法，可以多啟幾個(gè)線程

threading.Timer

既然第一種方法暴力，那么有沒有比較優(yōu)雅點(diǎn)的方法？
Python 標(biāo)準(zhǔn)庫 threading 中有個(gè) Timer 類。
它會(huì)新啟動(dòng)一個(gè)線程來執(zhí)行定時(shí)任務(wù)，所以它是非阻塞函式。

原理：線程中預(yù)置一個(gè)finished的事件，通過finished.wait等待固定時(shí)間間隔。
超時(shí)則執(zhí)行任務(wù)。如果需要取消任務(wù)，可以調(diào)用Timer.cancel來取消任務(wù)。

from threading import Timer

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

優(yōu)點(diǎn)：
不阻塞主進(jìn)程，task在線程中執(zhí)行

缺點(diǎn)：
一個(gè)Timer只能執(zhí)行一次task就結(jié)束了。
如果想循環(huán)，需要改造一下task函數(shù)。

from threading import Timer

def repeat_task():
    t = Timer(5, repeat_task)
    # 開始任務(wù)的位置決定了是任務(wù)之間等待固定間隔時(shí)間
    # 還是每個(gè)任務(wù)的開始等待固定間隔時(shí)間
    t.start()
    task()

這樣可以循環(huán)執(zhí)行，但是仍然只能一個(gè)線程一個(gè)任務(wù)。

Q：如何跳出循環(huán)
A：加入一個(gè)標(biāo)識(shí)符：

• sleep可以用一個(gè)布爾值來控制
• Timer可以用一個(gè)Event來控制

固定時(shí)間點(diǎn)執(zhí)行任務(wù)

以上是用內(nèi)置的方法中比較簡單的實(shí)現(xiàn)方式。簡單的功能可以實(shí)現(xiàn)。

前面執(zhí)行的都是指定間隔時(shí)間的定時(shí)任務(wù)，那怎么執(zhí)行指定時(shí)間點(diǎn)的任務(wù)呢？

上面的方法是可以做到的。有兩種思路，

• 前面我們指定了間隔時(shí)間，那指定時(shí)間點(diǎn)，就先計(jì)算當(dāng)前時(shí)間到指定時(shí)間點(diǎn)的間隔時(shí)間
• 不管間隔時(shí)間，而是記錄任務(wù)時(shí)間點(diǎn)，然后實(shí)時(shí)去檢查是否到達(dá)指定時(shí)間點(diǎn)

思路有了，但是對(duì)于固定時(shí)間點(diǎn)執(zhí)行任務(wù)的場景以及后面更復(fù)雜的場景，自己實(shí)現(xiàn)可能就變得復(fù)雜。

下面再介紹幾個(gè)進(jìn)階的定時(shí)任務(wù)的實(shí)現(xiàn)方式，可以適應(yīng)更復(fù)雜的業(yè)務(wù)場景。

sched

第三種方式是使用標(biāo)準(zhǔn)庫中sched模塊。sched 是事件調(diào)度器，
它通過 scheduler 類來調(diào)度事件，從而達(dá)到定時(shí)執(zhí)行任務(wù)的效果。

簡單示例如下：

import sched

schedule = sched.scheduler()  # 初始化 sched 模塊的 scheduler 類
schedule.enter(10, 1, task)  # 增加調(diào)度任務(wù)
schedule.run()  # 開始調(diào)度任務(wù)

scheduler 提供了兩個(gè)添加調(diào)度任務(wù)的函數(shù):

• enter(delay, priority, action, argument=(), kwargs={})該函數(shù)可以延遲一定時(shí)間執(zhí)行任務(wù)。delay 表示延遲多長時(shí)間執(zhí)行任務(wù)，單位是秒。
  priority為優(yōu)先級(jí)，越小優(yōu)先級(jí)越大。兩個(gè)任務(wù)指定相同的延遲時(shí)間，優(yōu)先級(jí)大的任務(wù)會(huì)向被執(zhí)行。
  action 即需要執(zhí)行的函數(shù)，argument 和 kwargs 分別是函數(shù)的位置和關(guān)鍵字參數(shù)。

• scheduler.enterabs(time, priority, action, argument=(), kwargs={})添加一項(xiàng)任務(wù)，但這個(gè)任務(wù)會(huì)在 time 這時(shí)刻執(zhí)行。因此，time 是絕對(duì)時(shí)間。其他參數(shù)用法與 enter() 中的參數(shù)用法是一致。

優(yōu)點(diǎn)：

• 執(zhí)行時(shí)間間隔和時(shí)間點(diǎn)執(zhí)行任務(wù)
• 可以添加不同的任務(wù)
• 任務(wù)可以設(shè)置優(yōu)先級(jí)

缺點(diǎn)：
scheduler 中的每個(gè)調(diào)度任務(wù)只會(huì)工作一次，不會(huì)無限循環(huán)被調(diào)用。如果想重復(fù)執(zhí)行同一任務(wù)， 需要重復(fù)添加調(diào)度任務(wù)即可。

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()

實(shí)現(xiàn)原理

當(dāng)然我們僅僅學(xué)會(huì)怎么用還是不夠的，不能知其然而不知其所以然。
介紹了那么多方法，那么多庫，但是底層的實(shí)現(xiàn)邏輯都是差不多的。
一個(gè)完整的定時(shí)任務(wù)系統(tǒng)，有三個(gè)部分：

• 任務(wù)隊(duì)列(task queue)
  根據(jù)執(zhí)行時(shí)間和優(yōu)先級(jí)進(jìn)行排序
  排序sorted(): 任務(wù)一多，整個(gè)隊(duì)列重排，性能堪憂
  heapq.pop()：堆排序，只獲取最高優(yōu)先級(jí)的任務(wù)，不重復(fù)排序
• 調(diào)度器(scheduler)
• 執(zhí)行器(executor)
  從前到后，實(shí)現(xiàn)的內(nèi)容越來越多，控制的精細(xì)度也就越來越高，實(shí)現(xiàn)的功能也就越來越豐富。

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):
    # 新建任務(wù)的前后，決定了是在任務(wù)結(jié)束后等待時(shí)間，還是固定間隔
    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)

到此這篇關(guān)于Python中實(shí)現(xiàn)定時(shí)任務(wù)詳解的文章就介紹到這了,更多相關(guān)Python中實(shí)現(xiàn)定時(shí)任務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論