設計模式是軟體開發中解決常見問題的經驗結晶。Python 動態語言特性讓某些模式的實作更為簡潔優雅。本文深入解析兩個行為型與建立型模式——備忘錄模式與物件池模式,提供可直接執行的完整範例,並說明適用場景與注意事項。
備忘錄模式在不破壞封裝性的前提下,捕獲物件內部狀態並在該物件之外儲存,以便日後恢復。經典應用場景:撤銷操作、交易回滾、表單狀態保存。
import copy
def Memento(obj, deep=False):
"""建立物件狀態快照,回傳恢復函式(閉包)"""
state = (copy.copy if deep else copy.deepcopy)(obj.__dict__)
def Restore():
obj.__dict__ = state
return Restore
class Transaction:
"""管理多個物件的交易:提交(儲存快照)與回滾(恢復快照)"""
deep = False
def __init__(self, *targets):
self.targets = targets
self.Commit()
def Commit(self):
"""建立所有目標物件的當前狀態快照"""
self.states = [Memento(target, self.deep) for target in self.targets]
def Rollback(self):
"""依序執行所有恢復函式,還原至上次 Commit 狀態"""
for state in self.states:
state()
def transactional(method):
"""裝飾器:為方法提供自動交易語意,異常時自動回滾"""
def wrapped(self, *args, **kwargs):
state = Memento(self)
try:
return method(self, *args, **kwargs)
except Exception:
state() # 發生異常,恢復狀態
raise
return wrapped
# === 示範類別 ===
class NumObj:
def __init__(self, value):
self.value = value
def __repr__(self):
return f'<{self.__class__.__name__}: {self.value!r}>'
def Increment(self):
self.value += 1
@transactional
def DoStuff(self):
# 故意製造錯誤:字串無法自增
self.value = '1111'
self.Increment() # TypeError
# === 測試流程 ===
if __name__ == '__main__':
n = NumObj(-1)
print(f'初始: {n}')
t = Transaction(n)
try:
for i in range(3):
n.Increment()
print(f'Increment {i+1}: {n}')
t.Commit() # 保存狀態:value = 2
print('-- committed')
for i in range(3):
n.Increment()
print(f'Increment {i+1}: {n}')
n.value += 'x' # 故意觸發 TypeError
print(f'錯誤操作後: {n}')
except Exception:
t.Rollback() # 回滾至 value = 2
print('-- rolled back')
print(f'回滾後: {n}')
print('-- now doing stuff ...')
try:
n.DoStuff() # 裝飾器版交易,異常自動回滾
except Exception:
print('-> doing stuff failed!')
import traceback; traceback.print_exc(0)
print(f'最終狀態: {n}')
初始:
Increment 1:
Increment 2:
Increment 3:
-- committed
Increment 1:
Increment 2:
Increment 3:
錯誤操作後:
-- rolled back
回滾後:
-- now doing stuff ...
-> doing stuff failed!
最終狀態:
Memento 回傳 Restore 函式,內部攜帶 state,呼叫端無需知道內部結構。deep 控制 copy.deepcopy 或 copy.copy,應對巢狀物件結構。@transactional 讓單一方法自動具備「異常回滾」能力。Transaction 可同時管理多個目標物件的一致性。記憶體開銷:若物件狀態龐大(如大型列表、圖片資料),頻繁建立快照會消耗大量記憶體。可考慮:
物件池預先建立一組可重用物件,使用者「借用」而非「新建」,用畢「歸還」而非「銷毀」。適用於:建立成本高(資料庫連線、執行緒、網路通訊端)、頻繁建立/銷毀導致 GC 壓力大的場景。
import queue
import threading
from contextlib import contextmanager
import types
class ObjectPool:
"""通用物件池:支援類別、函式建立物件,執行緒安全"""
def __init__(self, factory, *args, max_size=1, **kwargs):
"""
:param factory: 類別或函式,用於建立新物件
:param args: 傳遞給 factory 的位置參數
:param max_size: 池最大容量
:param kwargs: 傳遞給 factory 的關鍵字參數
"""
self.factory = factory
self.args = args
self.kwargs = kwargs
self.max_size = max_size
self._pool = queue.Queue()
self._lock = threading.Lock()
self._created = 0
def _create_object(self):
"""根據 factory 類型建立物件"""
if isinstance(self.factory, type): # 類別
return self.factory(*self.args, **self.kwargs)
elif callable(self.factory): # 函式
return self.factory(*self.args, **self.kwargs)
else:
raise TypeError("factory 必須是類別或可呼叫物件")
def borrow(self):
"""從池中取得物件,若池空且未達上限則建立新物件"""
with self._lock:
if self._pool.empty() and self._created < self.max_size:
obj = self._create_object()
self._created += 1
return obj
try:
return self._pool.get_nowait()
except queue.Empty:
# 池已滿且無可用物件:阻塞等待或拋出異常
return self._pool.get() # 阻塞直到有物件歸還
def return_obj(self, obj):
"""歸還物件至池中"""
if obj is not None:
self._pool.put(obj)
@contextmanager
def pooled(pool):
"""Context Manager 簡化借用/歸還流程,確保異常時也能歸還"""
obj = pool.borrow()
try:
yield obj
except Exception:
# 可選:標記物件損壞、重建等
raise
finally:
pool.return_obj(obj)
# === 測試:模擬高併發存取 ===
def echo_func(num):
return num
class EchoClass:
pass
if __name__ == '__main__':
# 建立兩個池:函式池、類別池
func_pool = ObjectPool(echo_func, 23, max_size=4)
cls_pool = ObjectPool(EchoClass, max_size=4)
class Worker(threading.Thread):
def run(self):
with pooled(func_pool) as f:
print(f'Func pool: {f}')
with pooled(cls_pool) as c:
print(f'Class pool: {c}')
threads = [Worker() for _ in range(200)]
for t in threads: t.start()
for t in threads: t.join(timeout=1.0)
print('測試完成')
| 特性 | 實作方式 |
|---|---|
| 執行緒安全 | threading.Lock 保護池狀態;queue.Queue 內建執行緒安全 |
| 支援類別與函式 | isinstance(factory, type) 區分,類別用實例化,函式用呼叫 |
| 最大容量控制 | max_size 限制建立總數,避免資源耗盡 |
| Context Manager | @contextmanager 裝飾器實作 pooled(),with 區塊自動歸還 |
| 阻塞等待 | 池滿時 queue.get() 阻塞,自然實現背壓 |
psycopg2.pool、SQLAlchemy QueuePool 皆為此模式。concurrent.futures.ThreadPoolExecutor 內建管理。urllib3.PoolManager、requests.Session 連線重用。finally 或 with 中歸還,否則池會永久縮小。備忘錄模式以閉包優雅實現狀態快照與恢復,適合撤銷、交易回滾場景;物件池模式透過預建與重用降低高成本物件的建立開銷,配合 Context Manager 確保資源正確循環。兩者皆展現 Python 動態特性帶來的實作簡潔性。實際專案中,建議優先使用成熟庫(如 sqlalchemy.pool、redis.ConnectionPool),僅在特殊需求下自行實作。