裝飾器簡單介紹

• 裝飾器是可調(diào)用的對象，其參數(shù)是另一個(gè)函數(shù)（被裝飾的函數(shù)）。 裝飾器可能會(huì)處理被裝飾的函數(shù)，然后把它返回，或者將其替換成另一個(gè)函數(shù)或可調(diào)用對象。下面看個(gè)簡單的例子：

@decorate
def target():
    pass

• 其實(shí)就是：

target = decorate(target)

• 裝飾器的一大特性是，能把被裝飾的函數(shù)替換成其他函數(shù)。第二個(gè)特性是，裝飾器在加載模塊時(shí)立即執(zhí)行。
• 也就是說，只要你給某個(gè)函數(shù)新增了裝飾器，這個(gè)函數(shù)就已經(jīng)通過自由變量的方式傳入給了裝飾器函數(shù)，這個(gè)函數(shù)內(nèi)存指向了裝飾器所在的函數(shù)。這里的加載時(shí)你可以理解為導(dǎo)入時(shí)，這樣就可以在函數(shù)運(yùn)行前做一些其他的操作。下面通過單例模式簡單分析下裝飾器。

裝飾器之單例模式分析

• 所謂單例模式，簡單來說就是類的實(shí)例只能存在一個(gè)。下面直接看代碼進(jìn)行分析：

from functools import wraps

def Singleton(cls):  # 傳入類（cls）而不是實(shí)例
    """單例模式之裝飾器實(shí)現(xiàn)"""
    instance_dict = {}  # 使用字典存儲(chǔ)類的實(shí)例

    @wraps(cls)  # 消除被裝飾函數(shù)內(nèi)置屬性和方法被替換的影響
    def wrapper(*args, **kwargs):  # 解包傳入?yún)?shù)
        if cls not in instance_dict: 
            # 如果類 cls 不在字典 instance_dict 的 key 中，則調(diào)用類cls構(gòu)造方法新建實(shí)例
            instance_dict[cls] = cls(*args, **kwargs)
        return instance_dict[cls]  # 返回類cls的實(shí)例

    return wrapper  # 返回閉包函數(shù)

• 通過代碼逐行分析應(yīng)該很清楚，只要給某個(gè)類添加了裝飾器，這個(gè)類在初始化時(shí)直接進(jìn)入裝飾器中的閉包函數(shù)返回類的唯一實(shí)例。
• 但是這個(gè)裝飾器還有個(gè)顯而易見的問題：線程不安全。當(dāng)有多個(gè)線程同時(shí)去獲取這個(gè)單例資源時(shí)，裝飾器是不會(huì)對線程進(jìn)行限制的。解決方法也很簡單，給裝飾器中的閉包函數(shù)加鎖即可，也就是說，該資源只能被一個(gè)線程訪問，只有該線程釋放了鎖，其他線程才能訪問。如下示例

import threading
from functools import wraps


def synchronized(func):
    '''線程鎖裝飾器'''
    func.__lock__ = threading.Lock()

    def syn_func(*args, **kwargs):
        with func.__lock__:
            return func(*args, **kwargs)

    return syn_func


def Singleton(cls):
    instance_dict = {}

    @synchronized
    @wraps(cls)
    def wrapper(*args, **kwargs):
        if cls not in instance_dict:
            instance_dict[cls] = cls(*args, **kwargs)
        return instance_dict[cls]

    return wrapper


@Singleton
class Foo:
    pass


if __name__ == '__main__':
    f1 = Foo()
    f2 = Foo()
    print(f1 is f2)  # True

• 給wrapper閉包函數(shù)加鎖，這樣無論有多少線程，只要沒有獲取鎖資源，都會(huì)進(jìn)行等待，直至上一個(gè)線程釋放鎖，這樣就解決了多線程下資源的安全獲取。
• 單例模式是個(gè)經(jīng)典的例子，下面步入正題，說說在實(shí)際項(xiàng)目中如何使用裝飾器

使用裝飾器進(jìn)行參數(shù)校驗(yàn)

• 在項(xiàng)目開發(fā)中進(jìn)行參數(shù)校驗(yàn)是個(gè)再正常不過得事情，畢竟我們無法保證前端傳入正確的數(shù)據(jù)。比如在 fastapi 中最常使用的就是 pydantic 庫，這個(gè)庫十分強(qiáng)大，使用起來也很簡單。但是這不是我們討論的主題，現(xiàn)在說下該如何使用裝飾器對傳入的參數(shù)進(jìn)行校驗(yàn)。
• 比如你想對傳入的參數(shù)進(jìn)行非空校驗(yàn)，但是又不想在函數(shù)里面調(diào)用其他函數(shù)進(jìn)行參數(shù)處理，使用裝飾器就可以很優(yōu)雅的實(shí)現(xiàn)，這其實(shí)就是設(shè)計(jì)模式的思想，AOP就是這樣做的。下面直接看代碼：

def para_none_check(func):
    """參數(shù)非空校驗(yàn)"""

    @wraps(func)
    def none_check(data):
        if not data:
            raise TypeError("傳入?yún)?shù)不能為空")
        return func(data)

    return none_check

• 下面來簡單測試下

@para_none_check
def func_test(data):
    print("func_test running")
   

if __name__ == '__main__':
    try:
        func_test([])
    except TypeError as e:
        print(e)  # 傳入?yún)?shù)不能為空

• 可知，當(dāng)傳入?yún)?shù)為空時(shí)拋出 TypeError 錯(cuò)誤，如果不想拋出錯(cuò)誤也可以，直接返回錯(cuò)誤提示也可以，比如把 raise TypeError("傳入?yún)?shù)不能為空") 替換為 return {"code": "01", "error": "傳入?yún)?shù)不能為空"}。
• 如果我想傳入指定的關(guān)鍵字參數(shù)，這又該如果做呢？比如我想傳入的參數(shù)里面必須包含 name和 age。你可能第一時(shí)間想到這樣限定：

def func(name, age, **kwargs):

• 但是實(shí)際情況很多時(shí)候往往無法直接獲取 name 和 age，因?yàn)閿?shù)據(jù)是從前端獲取的，我們只能拿到傳入的數(shù)據(jù)，并不知道這些數(shù)據(jù)是不是包含這些字段。不過有了前面的的基礎(chǔ)，現(xiàn)在對參數(shù)進(jìn)行校驗(yàn)也就不難了，示例如下：

def para_verification(required_fields: list):  # 指定要傳入的參數(shù)字段列表
    """參數(shù)校驗(yàn)，必須傳入裝飾器指定的參數(shù)"""

    def decorate(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            para = list(kwargs.keys())  # 獲取傳入的全部參數(shù)字段列表
            for field in required_fields:
                if field not in para:
                    return {"code": "01", "error": "傳入?yún)?shù)有誤", "required_fields": required_fields, 'kwargs': kwargs}
            return func(*args, **kwargs)

        return wrapper

    return decorate

• 如果難以理解的話先來看個(gè)測試示例：

@para_verification(['name', 'age'])  # 指定func_test必須傳入name和age
def func_test(*args, **kwargs):
    print("func_test running")
    return True


if __name__ == '__main__':
    kw1 = {"name": "張三", "height": 175}
    kw2 = {"name": "張三", "height": 175, "age": 18}
    print(func_test(**kw1))
    # {'code': '01', 'error': '傳入?yún)?shù)有誤', 'required_fields': ['name', 'age'], 'kwargs': {'name': '張三', 'height': 175}}
    print(func_test(**kw2))
    # func_test running
    # True

• 通過示例得出只要在裝飾器參數(shù)中指定要傳入的字段列表，實(shí)際運(yùn)行函數(shù)的時(shí)候會(huì)先用裝飾器中的參數(shù)和傳入的參數(shù)進(jìn)行校驗(yàn)，如果指定要傳入的參數(shù)都存在，則校驗(yàn)通過，直接運(yùn)行該函數(shù)。很多框架都利用了這樣的思想。
• 當(dāng)然，參數(shù)校驗(yàn)可不是這么簡單的事情，這里只是做了初步分析，實(shí)際場景可能遠(yuǎn)遠(yuǎn)比這復(fù)雜，但是萬變不離其中，你只要理解了裝飾器的原理，編碼自然水到渠成。
• 說了這么多只是為了讓你對裝飾器有更深的體會(huì)。而且舉出的實(shí)例都是很可能在實(shí)際項(xiàng)目中使用的，而且可以直接拿來使用。下面講講如何利用裝飾器以及一些模塊進(jìn)行內(nèi)存占用檢測。

接口內(nèi)存泄露檢測

• 可能有人會(huì)問，python作為一種動(dòng)態(tài)語言，也會(huì)存在內(nèi)存泄露嗎？pyhton確實(shí)不容易發(fā)生內(nèi)存泄露，但是并不表示不會(huì)發(fā)生。循環(huán)引用就是一個(gè)典型的例子，python解釋器會(huì)在對象的引用計(jì)數(shù)歸零時(shí)刪除該對象。
• 那到底什么是引用計(jì)數(shù)呢？比如a = [1, 2],給 a 賦值了一個(gè)列表對象，那么 a 就指向了這個(gè)列表的內(nèi)存地址，這稱為強(qiáng)引用，因?yàn)?code>弱應(yīng)用不太常見，所以強(qiáng)引用一般就稱為引用，如果我們再賦值 b = a, 這樣列表對象又有了一個(gè)引用，引用計(jì)數(shù)為2。如果我們這時(shí)候刪除a：del a，這樣其實(shí)是刪除了變量 a 對列表對象的引用，并沒有直接刪除列表這個(gè)對象。
• 簡單來說，就是所有指向?qū)ο蟮淖兞慷疾辉诖嬖诤蟛艜?huì)去刪除這個(gè)對象，接著上面，給b重新賦值改變b的內(nèi)存指向：b = [1, 4]，這時(shí)候列表對象 [1,2] 沒有任何變量指向它，引用計(jì)數(shù)歸零，該對象被gc刪除回收內(nèi)存。
• 簡單了解了python 的垃圾回收機(jī)制，我們也就知道為什么會(huì)出現(xiàn)內(nèi)存泄露了：只要一個(gè)對象的引用計(jì)數(shù)沒有歸零，那么這個(gè)對象就不會(huì)被刪除。但是實(shí)際項(xiàng)目中難的不是如何解決內(nèi)存泄露，而是如何檢測出哪里發(fā)生了內(nèi)存泄露。
• 檢測內(nèi)存泄露的方式有很多，比如使用pyrasite庫進(jìn)行遠(yuǎn)程檢測。其實(shí)內(nèi)存檢測是提前設(shè)置的，我們只需要在接口上添加裝飾器，然后在裝飾器中統(tǒng)計(jì)接口運(yùn)行時(shí)各個(gè)對象的占用內(nèi)存或者哪些文件的第幾行代碼占用內(nèi)存。這里使用python3 內(nèi)置的 tracemalloc 庫就可以了。

• 先說下tracemalloc 的用法，如下示例：

  
  

• 打印結(jié)果如下

  
  
• 可以看到第40行占用了3532kb內(nèi)存，這樣就能知道哪行代碼存在性能問題從而進(jìn)行優(yōu)化。

• 也可以在不同地方新建快照，比較代碼運(yùn)行后的性能差異：

  
  

• 測試打印結(jié)果

  
  
• 可以看到占到用較高的就只有第40行。
• 學(xué)會(huì)了tracemalloc 的基本使用，接下來使用裝飾器來實(shí)現(xiàn)：

import tracemalloc as tc
from functools import wraps
from loguru import logger

def interface_memory_leak_check(func):
    """接口內(nèi)存占用檢測"""

    @wraps(func)
    def wrapper(*args, **kwargs):
        tc.start()  # 開始跟蹤內(nèi)存分配
        snapshot1 = tc.take_snapshot()  # 建立快照
        re_data = func(*args, **kwargs)
        snapshot2 = tc.take_snapshot()  # 建立快照
        top_stats = snapshot2.compare_to(snapshot1, 'lineno')  # 比較兩段快照之間的內(nèi)存
        logger.info("--------------------[ Top 10 differences ]----------------------")
        for stat in top_stats[:10]:
            logger.info(stat)
        tc.stop()
        return re_data

    return wrapper

• 這里其實(shí)就是對接口運(yùn)行前后內(nèi)存占用進(jìn)行檢測，然后日志打印占用最高內(nèi)存的十個(gè)地方。

• 接下來我用實(shí)際項(xiàng)目接口進(jìn)行測試

  
  

• 然后使用測試工具測試該接口，我這里使用的yaki，一般來說postman也就足夠了。測試后看下docker容器部分日志如下所示：

  
  
• 可以看出占用內(nèi)存前十的基本都是第三方庫，說明該接口我們自己寫的代碼在性能這塊沒有出現(xiàn)大的問題。
• 當(dāng)然，這只是性能檢測的小試牛刀 ，內(nèi)存占用檢測很多實(shí)際情況是比較復(fù)雜的，但是這已經(jīng)足以說明裝飾器的強(qiáng)大了。
• 如果你完全沒有用過裝飾器，那現(xiàn)在開始還不晚，前提是你得知道裝飾器的原理，理解什么是自由變量和閉包。