裝飾器簡單介紹
-
裝飾器是可調用的對象,其參數(shù)是另一個函數(shù)(被裝飾的函數(shù))。 裝飾器可能會處理被裝飾的函數(shù),然后把它返回,或者將其替換成另一個函數(shù)或可調用對象。下面看個簡單的例子:
@decorate
def target():
pass
- 其實就是:
target = decorate(target)
- 裝飾器的一大特性是,
能把被裝飾的函數(shù)替換成其他函數(shù)。第二個特性是,裝飾器在加載模塊時立即執(zhí)行。 - 也就是說,只要你給某個函數(shù)新增了裝飾器,這個函數(shù)就已經通過
自由變量的方式傳入給了裝飾器函數(shù),這個函數(shù)內存指向了裝飾器所在的函數(shù)。這里的加載時你可以理解為導入時,這樣就可以在函數(shù)運行前做一些其他的操作。下面通過單例模式簡單分析下裝飾器。
裝飾器之單例模式分析
- 所謂單例模式,簡單來說就是類的實例只能存在一個。下面直接看代碼進行分析:
from functools import wraps
def Singleton(cls): # 傳入類(cls)而不是實例
"""單例模式之裝飾器實現(xiàn)"""
instance_dict = {} # 使用字典存儲類的實例
@wraps(cls) # 消除被裝飾函數(shù)內置屬性和方法被替換的影響
def wrapper(*args, **kwargs): # 解包傳入參數(shù)
if cls not in instance_dict:
# 如果類 cls 不在字典 instance_dict 的 key 中,則調用類cls構造方法新建實例
instance_dict[cls] = cls(*args, **kwargs)
return instance_dict[cls] # 返回類cls的實例
return wrapper # 返回閉包函數(shù)
- 通過代碼逐行分析應該很清楚,只要給某個類添加了裝飾器,這個類在初始化時直接進入裝飾器中的閉包函數(shù)返回類的唯一實例。
- 但是這個裝飾器還有個顯而易見的問題:
線程不安全。當有多個線程同時去獲取這個單例資源時,裝飾器是不會對線程進行限制的。解決方法也很簡單,給裝飾器中的閉包函數(shù)加鎖即可,也就是說,該資源只能被一個線程訪問,只有該線程釋放了鎖,其他線程才能訪問。如下示例
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ù)加鎖,這樣無論有多少線程,只要沒有獲取鎖資源,都會進行等待,直至上一個線程釋放鎖,這樣就解決了多線程下資源的安全獲取。
- 單例模式是個經典的例子,下面步入正題,說說在實際項目中如何使用裝飾器
使用裝飾器進行參數(shù)校驗
- 在項目開發(fā)中進行參數(shù)校驗是個再正常不過得事情,畢竟我們無法保證前端傳入正確的數(shù)據。比如在 fastapi 中最常使用的就是
pydantic庫,這個庫十分強大,使用起來也很簡單。但是這不是我們討論的主題,現(xiàn)在說下該如何使用裝飾器對傳入的參數(shù)進行校驗。 - 比如你想對傳入的參數(shù)進行非空校驗,但是又不想在函數(shù)里面調用其他函數(shù)進行參數(shù)處理,使用裝飾器就可以很優(yōu)雅的實現(xiàn),這其實就是設計模式的思想,AOP就是這樣做的。下面直接看代碼:
def para_none_check(func):
"""參數(shù)非空校驗"""
@wraps(func)
def none_check(data):
if not data:
raise TypeError("傳入參數(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) # 傳入參數(shù)不能為空
- 可知,當傳入參數(shù)為空時拋出 TypeError 錯誤,如果不想拋出錯誤也可以,直接返回錯誤提示也可以,比如把
raise TypeError("傳入參數(shù)不能為空")替換為return {"code": "01", "error": "傳入參數(shù)不能為空"}。 - 如果我想傳入指定的關鍵字參數(shù),這又該如果做呢?比如我想傳入的參數(shù)里面必須包含 name和 age。你可能第一時間想到這樣限定:
def func(name, age, **kwargs):
- 但是實際情況很多時候往往無法直接獲取 name 和 age,因為數(shù)據是從前端獲取的,我們只能拿到傳入的數(shù)據,并不知道這些數(shù)據是不是包含這些字段。不過有了前面的的基礎,現(xiàn)在對參數(shù)進行校驗也就不難了,示例如下:
def para_verification(required_fields: list): # 指定要傳入的參數(shù)字段列表
"""參數(shù)校驗,必須傳入裝飾器指定的參數(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": "傳入參數(shù)有誤", "required_fields": required_fields, 'kwargs': kwargs}
return func(*args, **kwargs)
return wrapper
return decorate
- 如果難以理解的話先來看個測試示例:
@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': '傳入參數(shù)有誤', 'required_fields': ['name', 'age'], 'kwargs': {'name': '張三', 'height': 175}}
print(func_test(**kw2))
# func_test running
# True
- 通過示例得出只要在裝飾器參數(shù)中指定要傳入的字段列表,實際運行函數(shù)的時候會先用裝飾器中的參數(shù)和傳入的參數(shù)進行校驗,如果指定要傳入的參數(shù)都存在,則校驗通過,直接運行該函數(shù)。很多框架都利用了這樣的思想。
- 當然,參數(shù)校驗可不是這么簡單的事情,這里只是做了初步分析,實際場景可能遠遠比這復雜,但是萬變不離其中,你只要理解了裝飾器的原理,編碼自然水到渠成。
- 說了這么多只是為了讓你對裝飾器有更深的體會。而且舉出的實例都是很可能在實際項目中使用的,而且可以直接拿來使用。下面講講如何利用裝飾器以及一些模塊進行內存占用檢測。
接口內存泄露檢測
- 可能有人會問,python作為一種動態(tài)語言,也會存在內存泄露嗎?pyhton確實不容易發(fā)生內存泄露,但是并不表示不會發(fā)生。循環(huán)引用就是一個典型的例子,
python解釋器會在對象的引用計數(shù)歸零時刪除該對象。 - 那到底什么是引用計數(shù)呢?比如
a = [1, 2],給 a 賦值了一個列表對象,那么 a 就指向了這個列表的內存地址,這稱為強引用,因為弱應用不太常見,所以強引用一般就稱為引用,如果我們再賦值b = a, 這樣列表對象又有了一個引用,引用計數(shù)為2。如果我們這時候刪除a:del a,這樣其實是刪除了變量 a 對列表對象的引用,并沒有直接刪除列表這個對象。 - 簡單來說,就是所有指向對象的變量都不在存在后才會去刪除這個對象,接著上面,給b重新賦值改變b的內存指向:
b = [1, 4],這時候列表對象 [1,2] 沒有任何變量指向它,引用計數(shù)歸零,該對象被gc刪除回收內存。 - 簡單了解了python 的垃圾回收機制,我們也就知道為什么會出現(xiàn)內存泄露了:
只要一個對象的引用計數(shù)沒有歸零,那么這個對象就不會被刪除。但是實際項目中難的不是如何解決內存泄露,而是如何檢測出哪里發(fā)生了內存泄露。 - 檢測內存泄露的方式有很多,比如使用pyrasite庫進行遠程檢測。其實內存檢測是提前設置的,我們只需要在接口上添加裝飾器,然后在裝飾器中統(tǒng)計接口運行時各個對象的占用內存或者哪些文件的第幾行代碼占用內存。這里使用python3 內置的 tracemalloc 庫就可以了。
-
先說下tracemalloc 的用法,如下示例:
-
打印結果如下
- 可以看到第40行占用了3532kb內存,這樣就能知道哪行代碼存在性能問題從而進行優(yōu)化。
-
也可以在不同地方新建快照,比較代碼運行后的性能差異:
-
測試打印結果
- 可以看到占到用較高的就只有第40行。
- 學會了tracemalloc 的基本使用,接下來使用裝飾器來實現(xiàn):
import tracemalloc as tc
from functools import wraps
from loguru import logger
def interface_memory_leak_check(func):
"""接口內存占用檢測"""
@wraps(func)
def wrapper(*args, **kwargs):
tc.start() # 開始跟蹤內存分配
snapshot1 = tc.take_snapshot() # 建立快照
re_data = func(*args, **kwargs)
snapshot2 = tc.take_snapshot() # 建立快照
top_stats = snapshot2.compare_to(snapshot1, 'lineno') # 比較兩段快照之間的內存
logger.info("--------------------[ Top 10 differences ]----------------------")
for stat in top_stats[:10]:
logger.info(stat)
tc.stop()
return re_data
return wrapper
- 這里其實就是對接口運行前后內存占用進行檢測,然后日志打印占用最高內存的十個地方。
-
接下來我用實際項目接口進行測試
-
然后使用測試工具測試該接口,我這里使用的yaki,一般來說postman也就足夠了。測試后看下docker容器部分日志如下所示:
- 可以看出占用內存前十的基本都是第三方庫,說明該接口我們自己寫的代碼在性能這塊沒有出現(xiàn)大的問題。
- 當然,這只是性能檢測的小試牛刀 ,內存占用檢測很多實際情況是比較復雜的,但是這已經足以說明裝飾器的強大了。
- 如果你完全沒有用過裝飾器,那現(xiàn)在開始還不晚,前提是你得知道裝飾器的原理,理解什么是自由變量和閉包。





