策略模式和觀察者模式是軟體開發中常用的兩種行為型設計模式。策略模式允許在執行時動態切換演算法，而觀察者模式則實作了物件狀態變化時的自動通知機制。在實際應用中，策略模式常與工廠模式結合使用，以簡化策略物件的建立過程。Lambda 表示式則提供了更簡潔的策略定義方式，適用於簡單的演算法場景。策略鏈則能組合多個策略，實作更複雜的業務邏輯。觀察者模式在處理物件狀態變化時非常有效，但需要注意執行緒安全和記憶體管理問題。使用鎖機制可以確保多執行緒環境下的資料一致性，而弱參照則能避免潛在的記憶體洩漏風險，提升應用程式的穩定性和效能。
策略模式（Strategy Pattern）的高階應用與實踐
前言
策略模式是一種行為設計模式，它允許在執行時選擇不同的演算法或行為來執行特定任務。透過將演算法封裝在獨立的策略類別中，開發者可以在不修改客戶端程式碼的情況下更換或擴充套件演算法。本文將探討策略模式的高階應用，包括其與其他設計模式的結合、效能最佳化、錯誤處理和安全考量。
策略模式的基本實作
首先，我們來看一個簡單的策略模式實作範例。在這個例子中，我們定義了一個 Strategy 抽象基礎類別和兩個具體策略類別 ConcreteStrategyA 和 ConcreteStrategyB。Context 類別負責委派演算法決策給策略物件。
from abc import ABC, abstractmethod
class Strategy(ABC):
@abstractmethod
def execute(self, data):
pass
class ConcreteStrategyA(Strategy):
def execute(self, data):
# 實作演算法 A 的細節
result = sorted(data)
return result
class ConcreteStrategyB(Strategy):
def execute(self, data):
# 實作演算法 B 的細節
result = list(reversed(sorted(data)))
return result
class Context:
def __init__(self, strategy: Strategy):
self._strategy = strategy
def set_strategy(self, strategy: Strategy):
self._strategy = strategy
def do_something(self, data):
return self._strategy.execute(data)
# 客戶端程式碼展示可互換性
data = [5, 2, 9, 1]
context = Context(ConcreteStrategyA())
print("策略 A:", context.do_something(data))
context.set_strategy(ConcreteStrategyB())
print("策略 B:", context.do_something(data))
內容解密：
Strategy抽象基礎類別：定義了execute抽象方法，強制所有具體策略類別實作此方法。ConcreteStrategyA和ConcreteStrategyB：實作了不同的排序演算法，分別進行升序和降序排序。Context類別：持有一個Strategy物件的參照，並將演算法執行委派給該策略物件。- 客戶端程式碼：展示瞭如何在執行時更換策略物件以改變
Context的行為。
與工廠模式結合
為了進一步解耦策略例項的建立過程，可以將策略模式與工廠模式結合使用。工廠模式負責根據組態或執行時條件動態建立策略例項。
class StrategyFactory:
_strategies = {
'ascend': ConcreteStrategyA,
'descend': ConcreteStrategyB,
}
@classmethod
def get_strategy(cls, key: str) -> Strategy:
strategy = cls._strategies.get(key)
if not strategy:
raise ValueError(f"策略 {key} 未找到")
return strategy()
# 使用工廠的客戶端程式碼
user_choice = 'ascend' # 可由使用者輸入或組態驅動
strategy_instance = StrategyFactory.get_strategy(user_choice)
context = Context(strategy_instance)
result = context.do_something(data)
print("結果:", result)
內容解密：
StrategyFactory類別：維護一個策略類別的對映表，並提供get_strategy方法根據鍵值取得對應的策略例項。- 動態建立策略：客戶端程式碼根據使用者選擇或組態動態取得策略例項，並將其傳入
Context中執行。
使用 Lambda 表示式作為策略
在某些情況下，演算法可能非常簡潔，此時可以使用 lambda 表示式或函式指標作為策略，以簡化設計。
class ContextLambda:
def __init__(self, strategy):
self._strategy = strategy
def set_strategy(self, strategy):
self._strategy = strategy
def do_something(self, data):
return self._strategy(data)
# 使用 lambda 表示式的策略實作
strategy_a = lambda data: sorted(data)
strategy_b = lambda data: list(reversed(sorted(data)))
context_lambda = ContextLambda(strategy_a)
print("Lambda 策略 A:", context_lambda.do_something(data))
context_lambda.set_strategy(strategy_b)
print("Lambda 策略 B:", context_lambda.do_something(data))
內容解密：
ContextLambda類別：與Context類別似，但接受 lambda 表示式或函式作為策略。- lambda 策略：使用 lambda 表示式實作簡單的排序演算法，並傳入
ContextLambda中執行。
策略鏈（Strategy Chain）
在複雜系統中，可能需要組合多個策略物件以實作更複雜的功能。策略鏈是一種常見的模式，其中一個策略的輸出成為另一個策略的輸入。
class StrategyChain(Strategy):
def __init__(self, strategies):
self._strategies = strategies
def execute(self, data):
temp = data
for strat in self._strategies:
temp = strat.execute(temp)
return temp
# 連結策略：先排序後反轉
chain_strategy = StrategyChain([ConcreteStrategyA(), ConcreteStrategyB()])
context_chain = Context(chain_strategy)
print("策略鏈結果:", context_chain.do_something(data))
內容解密：
StrategyChain類別：實作了一個策略鏈，依次執行多個策略。- 鏈式執行：首先對資料進行排序，然後反轉排序結果。
效能最佳化與快取
在效能敏感的應用中，需要考慮動態分派的開銷。透過 JIT 編譯技術或使用如 PyPy 的專門直譯器，可以最佳化方法呼叫並減少策略模式的開銷。此外，快取策略結果可以避免重複計算，提高系統吞吐量。
錯誤處理與安全考量
在根據策略的架構中，錯誤處理和安全考量至關重要。需要對策略執行進行強壯的例外處理，並實施輸入驗證和邊界檢查以防止惡意演算法注入。
觀察者模式：監控物件狀態變化
觀察者模式建立了一種強大的訂閱機制，多個物件（觀察者）註冊以接收有關某個主題狀態變化的更新。這種解耦使得主題可以在不知道觀察者的情況下廣播狀態變化，從而實作了靈活的設計，其中新的觀察者可以無縫地加入，而現有的觀察者可以在不影響整體系統架構的情況下被移除。在複雜系統中，狀態一致性和即時回應性至關重要，觀察者模式是有效傳播變更的主要策略。
核心實作
觀察者模式的核心是事件分派過程的抽象化。主題不僅維護內部狀態，還維護了對特定狀態變更感興趣的觀察者登入檔。當主題狀態被改變時，它會遍歷已註冊的觀察者，並透過標準化介面通知每個觀察者。實作方式可能有所不同，從同步回呼到非同步訊息傳遞系統，其中高階開發人員經常利用平行程式設計正規化。
Python 基礎實作範例
在 Python 中，典型的實作使用抽象基底類別來定義觀察者介面，每個觀察者實作一個專門的更新方法。主題維護了一個觀察者列表（或集合），並在狀態變更時呼叫每個觀察者的更新方法。以下是基礎實作範例：
from abc import ABC, abstractmethod
class Observer(ABC):
@abstractmethod
def update(self, subject):
pass
class Subject:
def __init__(self):
self._observers = set()
self._state = None
def attach(self, observer: Observer):
self._observers.add(observer)
def detach(self, observer: Observer):
self._observers.discard(observer)
def notify(self):
for observer in self._observers:
observer.update(self)
def set_state(self, state):
self._state = state
self.notify()
def get_state(self):
return self._state
class ConcreteObserver(Observer):
def update(self, subject: Subject):
print(f"{self.__class__.__name__} 觀察到狀態：{subject.get_state()}")
# 使用範例：
subject = Subject()
observer1 = ConcreteObserver()
observer2 = ConcreteObserver()
subject.attach(observer1)
subject.attach(observer2)
subject.set_state("初始化")
內容解密：
- 定義抽象觀察者類別：
Observer類別定義了抽象方法update，強制所有具體觀察者實作此方法以接收主題的更新通知。 - 主題類別實作：
Subject類別負責管理觀察者列表，並在狀態變更時通知所有註冊的觀察者。 - 註冊與通知機制：
attach和detach方法允許觀察者註冊或取消註冊，而notify方法遍歷所有註冊的觀察者並呼叫其update方法。 - 具體觀察者的實作：
ConcreteObserver類別是Observer的具體實作，當接收到主題的更新通知時，它會列印出當前狀態。
高階實作考量
執行緒安全
在多執行緒環境中，必須保護主題的觀察者登入檔免受平行修改。使用鎖或其他更高階別的平行控制機制是必要的。以下是一個執行緒安全的範例：
import threading
from abc import ABC, abstractmethod
class ThreadSafeSubject:
def __init__(self):
self._observers = set()
self._state = None
self._lock = threading.Lock()
def attach(self, observer: 'Observer'):
with self._lock:
self._observers.add(observer)
def detach(self, observer: 'Observer'):
with self._lock:
self._observers.discard(observer)
def notify(self):
with self._lock:
observers = list(self._observers)
for observer in observers:
observer.update(self)
def set_state(self, state):
with self._lock:
self._state = state
self.notify()
def get_state(self):
with self._lock:
return self._state
class Observer(ABC):
@abstractmethod
def update(self, subject: 'ThreadSafeSubject'):
pass
內容解密：
- 使用鎖保護分享資源：在
ThreadSafeSubject中，使用threading.Lock來確保對觀察者集合和狀態的存取是執行緒安全的。 - 最小化鎖持有時間：在
notify方法中，先在鎖的保護下建立觀察者列表的副本，然後釋放鎖，最後遍歷該列表以通知觀察者，以減少鎖的持有時間。
弱參照管理記憶體
在具有自動垃圾回收功能的語言（如 Python）中，如果主題持有對觀察者的強參照，可能會導致記憶體洩漏。使用弱參照可以解決這個問題。Python 的 weakref 模組允許主題維護一個弱參照集合，當觀察者不再被使用時，自動清除對其的參照。
import weakref
from abc import ABC, abstractmethod
class Observer(ABC):
@abstractmethod
def update(self, subject):
pass
class WeakSubject:
def __init__(self):
self._observers = weakref.WeakSet()
self._state = None
def attach(self, observer: Observer):
self._observers.add(observer)
def detach(self, observer: Observer):
self._observers.discard(observer)
def notify(self):
for observer in self._observers:
observer.update(self)
def set_state(self, state):
self._state = state
self.notify()
def get_state(self):
return self._state
內容解密：
- 使用弱參照集合：
WeakSubject使用weakref.WeakSet來儲存觀察者，這樣當觀察者不再被其他地方參照時，可以自動被垃圾回收。 - 避免記憶體洩漏：透過使用弱參照，即使主題仍然存在，觀察者也可以在不再被需要時被垃圾回收，從而避免了記憶體洩漏。
