任務超時退出

我們日常在使用的各種網絡請求庫時都帶有timeout參數，例如request庫。這個參數可以使請求超時就不再繼續了，直接拋出超時錯誤，避免等太久。

如果我們自己開發的方法也希望增加這個功能，該如何做呢？

方法很多，但最簡單直接的是使用并發庫futures，為了使用方便，我將其封裝成了一個裝飾器，代碼如下：

import functools
from concurrent import futures

executor = futures.ThreadPoolExecutor(1)

def timeout(seconds):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kw):
            future = executor.submit(func, *args, **kw)
            return future.result(timeout=seconds)
        return wrapper
    return decorator

定義了以上函數，我們就有了一個超時結束的裝飾器，下面可以測試一下：

import time

@timeout(1)
def task(a, b):
    time.sleep(1.2)
    return a+b

task(2, 3)

結果：

---------------------------------------------------------------------------
TimeoutError                              Traceback (most recent call last)
...
D:Anaconda3libconcurrentfutures_base.py in result(self, timeout)
    432                 return self.__get_result()
    433             else:
-- > 434                 raise TimeoutError()
    435 
    436     def exception(self, timeout=None):

TimeoutError:

上面我們通過裝飾器定義了函數的超時時間為1秒，通過睡眠模擬函數執行超過1秒時，成功的拋出了超時異常。

程序能夠在超時時間內完成時：

@timeout(1)
def task(a, b):
    time.sleep(0.9)
    return a+b

task(2, 3)

結果：

5

可以看到，順利的得到了結果。

這樣我們就可以通過一個裝飾器給任何函數增加超時時間，這個函數在規定時間內還處理不完就可以直接結束任務。

前面我將這個裝飾器將所需的變量定義到了外部，其實我們還可以通過類裝飾器進一步封裝，代碼如下：

import functools
from concurrent import futures

class timeout:
    __executor = futures.ThreadPoolExecutor(1)

    def __init__(self, seconds):
        self.seconds = seconds

    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kw):
            future = timeout.__executor.submit(func, *args, **kw)
            return future.result(timeout=self.seconds)
        return wrapper

經測試使用類裝飾器能得到同樣的效果。

注意：使用@functools.wraps的目的是因為被裝飾的func函數元信息會被替換為wrapper函數的元信息，而@functools.wraps(func)將wrapper函數的元信息替換為func函數的元信息。最終雖然返回的是wrapper函數，元信息卻依然是原有的func函數。

在函數式編程中，函數的返回值是函數對象被稱為閉包。

日志記錄

如果我們需要記錄部分函數的執行時間，函數執行前后打印一些日志，裝飾器是一種很方便的選擇。

代碼如下：

import time
import functools

def log(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        res = func(*args, **kwargs)
        end = time.perf_counter()
        print(f'函數 {func.__name__} 耗時 {(end - start) * 1000} ms')
        return res
    return wrapper

裝飾器 log 記錄某個函數的運行時間，并返回其執行結果。

測試一下：

@log
def now():
    print('2021-7-1')

now()

結果：

2021-7-1
函數 now 耗時 0.09933599994838005 ms

緩存

如果經常調用一個函數，而且參數經常會產生重復，如果把結果緩存起來，下次調用同樣參數時就會節省處理時間。

定義函數：

import math
import random
import time


def task(x):
    time.sleep(0.01)
    return round(math.log(x**3 / 15), 4)

執行：

%%time
for i in range(500):
    task(random.randrange(5, 10))

結果：

Wall time: 5.01 s

此時如果我們使用緩存的效果就會大不一樣，實現緩存的裝飾器有很多，我就不重復造輪子了，這里使用functools包下的LRU緩存：

from functools import lru_cache

@lru_cache()
def task(x):
    time.sleep(0.01)
    return round(math.log(x**3 / 15), 4)

執行：

%%time
for i in range(500):
    task(random.randrange(5, 10))

結果：

Wall time: 50 ms

約束某個函數的可執行次數

如果我們希望程序中的某個函數在整個程序的生命周期中只執行一次或N次，可以寫一個這樣的裝飾器：

import functools


class allow_count:
    def __init__(self, count):
        self.count = count
        self.i = 0

    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kw):
            if self.i >= self.count:
                return
            self.i += 1
            return func(*args, **kw)
        return wrapper

測試：

@allow_count(3)
def job(x):
    x += 1
    return x


for i in range(5):
    print(job(i))

結果：

1
2
3
None
None