Python裝飾器實作執行緒安全快取與重試機制

– 8 分鐘閱讀

在多執行緒環境下，共用資源的存取控制至關重要。本文將示範如何使用 Python 裝飾器實作執行緒安全的快取機制，避免競爭條件導致的資料錯誤。同時，我們也將探討如何使用裝飾器實作重試機制，提升程式碼的容錯能力。最後，我們將結合 Facade 模式，展示如何簡化對複雜系統的存取，並整合日誌記錄功能，方便監控和除錯。這些技術都能有效提升程式碼的可讀性、可維護性和執行效率。

執行緒安全快取裝飾器

為了確保在多執行緒環境中快取的安全存取，我們可以使用執行緒鎖（threading.Lock）來保護快取的存取。以下是實作執行緒安全快取裝飾器的示例：

import threading
import functools

def thread_safe_caching_decorator(func):
    cache = {}
    lock = threading.Lock()

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        key = (args, tuple(sorted(kwargs.items())))
        with lock:
            if key not in cache:
                cache[key] = func(*args, **kwargs)
        return cache[key]
    return wrapper

這個裝飾器使用一個鎖來保護快取的存取，確保只有一個執行緒可以存取快取同時。

例子：計算函式的執行緒安全快取

import time
import logging

def logging_decorator(func):
    def wrapper(*args, **kwargs):
        logging.info(f"Calling {func.__name__} with args {args} and kwargs {kwargs}")
        return func(*args, **kwargs)
    return wrapper

@logging_decorator
@thread_safe_caching_decorator
def compute_thread_safe(x, y):
    time.sleep(0.05)
    return x + y

print(compute_thread_safe(5, 10))

這個例子展示瞭如何使用執行緒安全快取裝飾器來保護計算函式的快取。

高階主題：裝飾器繼承和後設資料保留

當裝飾器應用於類別時，需要特別小心地處理類別的後設資料，例如 __name__、__doc__ 和方法解析順序（MRO）。在 Python 中，functools.wraps 裝飾器可以幫助維持屬性的一致性。

重試邏輯裝飾器

為了實作重試邏輯，可以建立一個裝飾器來攔截函式呼叫的異常，並根據需要重試。以下是示例：

def retry_decorator(max_retries=3, delay=0.1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"ATTEMPT {attempts+1} failed: {e}")
                    attempts += 1
                    time.sleep(delay)
        return wrapper
    return decorator

這個裝飾器可以用來保護函式呼叫免受暫時性失敗的影響，例如網路請求或使用者介面操作。

使用 Python 實作重試裝飾器和簡化複雜系統的 Facade 模式

1. 重試裝飾器（Retry Decorator）

在進行不穩定操作時，使用重試裝飾器可以增加程式的健壯性。以下是使用 Python 實作的重試裝飾器範例：

import time
from functools import wraps

def retry_decorator(max_retries=3, delay=0.2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    time.sleep(delay)
                    if attempts == max_retries:
                        raise Exception(f"Function {func.__name__} failed after {max_retries} retries: {str(e)}")
        return wrapper
    return decorator

@retry_decorator(max_retries=5, delay=0.2)
def unstable_operation(x):
    if x % 2 == 0:
        return f"Processed value {x}"
    else:
        raise ValueError("Unlucky odd value")

try:
    print(unstable_operation(3))
except Exception as error:
    print(error)

2. 簡化複雜系統的 Facade 模式

Facade 模式提供了一種簡單的方式來簡化複雜系統的使用。以下是使用 Python 實作的 Facade 模式範例：

class AuthenticationService:
    def authenticate(self, username, password):
        # Complex authentication logic with token management and encryption.
        if username == "admin" and password == "secret":
            return "token_abc123"
        raise ValueError("Authentication failed")

class DataService:
    def fetch_data(self, token):
        # Validate token and retrieve data from a remote database.
        if token == "token_abc123":
            return {"data": [1, 2, 3]}
        raise ValueError("Invalid token for data service")

class LoggingService:
    def log_event(self, message):
        # Log the event.
        print(f"Logged event: {message}")

class Facade:
    def __init__(self):
        self.auth_service = AuthenticationService()
        self.data_service = DataService()
        self.logging_service = LoggingService()

    def perform_complex_operation(self, username, password):
        try:
            token = self.auth_service.authenticate(username, password)
            data = self.data_service.fetch_data(token)
            self.logging_service.log_event(f"Data fetched successfully: {data}")
            return data
        except Exception as e:
            self.logging_service.log_event(f"Error occurred: {str(e)}")
            raise

# 使用Facade模式簡化複雜系統的使用。
facade = Facade()
try:
    result = facade.perform_complex_operation("admin", "secret")
    print(result)
except Exception as error:
    print(error)

高效率日誌記錄與遠端伺服器整合

在許多應用中，尤其是那些需要高吞吐量的日誌記錄系統時，能夠有效地記錄事件至遠端伺服器的能力是非常重要的。這不僅有助於監控和除錯，也能夠提供系統執行狀態的寶貴資訊。

服務外觀模式（Facade Pattern）

為了簡化複雜系統的存取和使用，服務外觀模式是一種非常有用的設計模式。它提供了一個統一的介面，使得客戶端能夠更容易地與多個子系統進行互動。

class ServiceFacade:
    def __init__(self):
        self.auth_service = AuthenticationService()
        self.data_service = DataService()
        self.log_service = LoggingService()

    def get_processed_data(self, username, password):
        try:
            # 進行身份驗證
            token = self.auth_service.authenticate(username, password)
            self.log_service.log_event(f"使用者 '{username}' 身份驗證成功")

            # 取得必要的資料
            data = self.data_service.fetch_data(token)
            self.log_service.log_event(f"使用者 '{username}' 資料取得成功")

            # 處理資料：可以包含高階別的資料操作或業務邏輯
            processed_data = self._process_data(data)

            return processed_data
        except Exception as e:
            self.log_service.log_event(f"發生錯誤：{e}")
            raise

    def _process_data(self, data):
        try:
            # 示例：非平凡的外觀內資料處理，例如聚合或轉換
            return sum(data["data"]) / len(data["data"])
        except Exception as e:
            raise ValueError("資料處理錯誤") from e

# 使用示範：
facade = ServiceFacade()
result = facade.get_processed_data("admin", "secret")

日誌記錄實作

在上述程式碼中，LoggingService類別負責處理日誌記錄。為了實作高效率的日誌記錄，特別是當需要記錄到遠端伺服器時，可以考慮使用佇列（Queue）機制或非同步日誌記錄方式，以避免阻塞主執行緒。

import logging
from queue import Queue
import threading

class LoggingService:
    def __init__(self):
        self.log_queue = Queue()
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.INFO)

        # 建立一個 handler 並設定級別為 INFO
        handler = logging.StreamHandler()
        handler.setLevel(logging.INFO)

        # 建立一個 formatter 並將其新增到 handler
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)

        # 將 handler 新增到 logger
        self.logger.addHandler(handler)

        # 啟動一個執行緒用於處理日誌佇列
        threading.Thread(target=self.process_log_queue).start()

    def log_event(self, message):
        self.log_queue.put(message)

    def process_log_queue(self):
        while True:
            message = self.log_queue.get()
            self.logger.info(message)
            self.log_queue.task_done()

# 使用示範：
log_service = LoggingService()
log_service.log_event("測試日誌記錄")

服務外觀模式（Facade Pattern）及其應用

服務外觀模式是一種結構型設計模式，讓客戶端能夠以簡單的方式與複雜的子系統進行互動。它提供了一個統一的介面，封裝了子系統中各個類別的複雜互動，讓客戶端不需要了解子系統的細節就能夠使用它。

服務外觀模式的優點

  1. 簡化客戶端程式碼：服務外觀模式簡化了客戶端程式碼，客戶端不需要了解子系統的複雜互動就能夠使用它。
  2. 提高子系統的獨立性：服務外觀模式使得子系統可以獨立演化，客戶端不需要了解子系統的細節就能夠使用它。
  3. 提高系統的可維護性：服務外觀模式使得系統更容易維護，客戶端不需要了解子系統的細節就能夠使用它。

服務外觀模式的實作

以下是服務外觀模式的一個實作範例：

import time

class ServiceFacade:
    def __init__(self):
        self.auth_service = AuthService()
        self.data_service = DataService()
        self.log_service = LogService()

    def get_processed_data(self, username, password):
        token = self.auth_service.authenticate(username, password)
        data = self.data_service.fetch_data(token)
        processed_data = self._process_data(data)
        return processed_data

    def _process_data(self, data):
        # 處理資料的邏輯
        pass

class CachedServiceFacade(ServiceFacade):
    def __init__(self):
        super().__init__()
        self._cache = {}
        self._cache_expiry = {}

    def get_processed_data(self, username, password):
        cache_key = f"{username}:{password}"
        current_time = time.time()

        # 使用快取的token如果有效
        if (cache_key in self._cache and
            current_time < self._cache_expiry.get(cache_key, 0)):
            token = self._cache[cache_key]
            self.log_service.log_event(f"使用快取的token для '{username}'")
        else:
            token = self.auth_service.authenticate(username, password)
            self._cache[cache_key] = token
            self._cache_expiry[cache_key] = current_time + 300  # 快取5分鐘
            self.log_service.log_event(f"新token快取 для '{username}'")

        try:
            data = self.data_service.fetch_data(token)
            processed_data = self._process_data(data)
            return processed_data
        except Exception as e:
            self.log_service.log_event(f"錯誤在快取的facade: {e}")
            raise

# 使用快取的facade
cached_facade = CachedServiceFacade()
result = cached_facade.get_processed_data("username", "password")
print(f"處理結果: {result}")

服務外觀模式的應用場景

  1. 複雜系統的簡化：服務外觀模式可以簡化複雜系統的互動，讓客戶端不需要了解子系統的細節就能夠使用它。
  2. 提高系統的可維護性：服務外觀模式可以提高系統的可維護性，客戶端不需要了解子系統的細節就能夠使用它。
  3. 提高系統的效能：服務外觀模式可以提高系統的效能，透過快取機制可以減少對子系統的請求。

3.5 Facade 模式：簡化複雜系統的介面

Facade 模式是一種結構模式，提供了一種簡化複雜系統的介面方式。它可以將多個子系統或模組封裝成一個統一的介面，讓客戶端可以方便地存取和使用這些子系統或模組。

Facade 模式的優點

  • 簡化複雜系統的介面：Facade 模式可以將多個子系統或模組封裝成一個統一的介面，讓客戶端可以方便地存取和使用這些子系統或模組。
  • 提高系統的可維護性：Facade 模式可以將子系統或模組的修改和擴充套件限制在 Facade 內，減少對客戶端的影響。
  • 提高系統的可測試性：Facade 模式可以方便地測試子系統或模組的功能，減少測試的複雜性。

Facade 模式的實作

class CachedServiceFacade:
    def __init__(self):
        self.cache = {}

    def get_processed_data(self, username, password):
        if (username, password) in self.cache:
            return self.cache[(username, password)]
        else:
            # 對子系統進行身份驗證和授權
            data = self.authenticate_and_authorize(username, password)
            self.cache[(username, password)] = data
            return data

    def authenticate_and_authorize(self, username, password):
        # 對子系統進行身份驗證和授權
        #...
        return "已經驗證和授權的資料"

cached_facade = CachedServiceFacade()
print(cached_facade.get_processed_data("admin", "secret"))

Facade 模式的應用場景

  • 簡化複雜系統的介面：Facade 模式可以將多個子系統或模組封裝成一個統一的介面，讓客戶端可以方便地存取和使用這些子系統或模組。
  • 提高系統的可維護性：Facade 模式可以將子系統或模組的修改和擴充套件限制在 Facade 內，減少對客戶端的影響。
  • 提高系統的可測試性：Facade 模式可以方便地測試子系統或模組的功能，減少測試的複雜性。

3.6 Flyweight 模式：分享物件以提高效率

Flyweight 模式是一種結構模式，提供了一種分享物件以提高效率的方式。它可以將多個物件中的共同狀態提取出來，形成一個獨立的物件，這樣可以減少物件的數量，提高效率。

Flyweight 模式的優點

  • 提高效率：Flyweight 模式可以減少物件的數量，提高效率。
  • 節省記憶體：Flyweight 模式可以節省記憶體，減少物件的數量。

Flyweight 模式的實作

class Glyph:
    def __init__(self, char, font, style):
        self.char = char
        self.font = font
        self.style = style

    def render(self, x, y, color):
        print(f"渲染 '{self.char}' 使用 {self.font} ({self.style}) 在位置 ({x}, {y}) 與顏色 {color}")

class GlyphFactory:
    _flyweights = {}

    @classmethod
    def get_flyweight(cls, char, font, style):
        key = (char, font, style)
        if key not in cls._flyweights:
            cls._flyweights[key] = Glyph(char, font, style)
        return cls._flyweights[key]

# 使用方法：建立多個字元的字形。
text = "HELLO FLYWEIGHT"
font = "Arial"
for char in text:
    glyph = GlyphFactory.get_flyweight(char, font, "normal")
    glyph.render(0, 0, "black")

Flyweight 模式的應用場景

  • 分享物件：Flyweight 模式可以將多個物件中的共同狀態提取出來，形成一個獨立的物件，這樣可以減少物件的數量，提高效率。
  • 節省記憶體：Flyweight 模式可以節省記憶體，減少物件的數量。

從系統架構的視角來看，本文探討的 Python 裝飾器，巧妙地利用了程式碼封裝和元程式設計的特性，為程式碼的重用性和可維護性提供了優雅的解決方案。執行緒安全快取裝飾器有效地解決了多執行緒環境下的資料一致性問題，但需要注意鎖的粒度控制，避免過度鎖定影響效能。重試裝飾器則提升了程式碼的容錯能力，但需謹慎設定重試次數和延遲時間，避免無限重試或資源耗盡。 Facade 模式則透過提供簡化的介面，降低了系統的複雜度，提升了程式碼的可讀性和可維護性，但也可能隱藏底層系統的細節，增加除錯的難度。隨著 Python 語言和相關技術的發展，裝飾器和設計模式的應用將更加廣泛和深入，例如非同步程式設計和分散式系統的場景。對於開發者而言，深入理解這些技術的核心概念和最佳實踐，將有助於構建更具彈性、高效和可維護的軟體系統。因此，建議開發者在實務中積極探索並應用這些技術，持續精程式式設計的技藝。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。