返回文章列表

效能設計模式提升應用程式速度

本文深入探討效能設計模式,例如快取策略(Cache-Aside)、記憶化(Memoization)和延遲載入(Lazy Loading),並提供 Python

軟體開發 效能最佳化

在現今軟體開發中,效能最佳化至關重要。除了平行和非同步模式外,效能設計模式也扮演著關鍵角色,例如快取策略、記憶化和延遲載入。這些模式提供有效方法來提升應用程式速度、減少記憶體使用並增強可擴充套件性。本文將著重探討快取策略模式(Cache-Aside)和記憶化模式(Memoization),並提供實際的 Python 程式碼範例,示範如何應用這些模式來最佳化資料函式庫讀取和函式呼叫的效能。同時,我們也會探討延遲載入的應用場景,並提供程式碼範例。

效能設計模式

在前一章中,我們討論了平行和非同步模式,這些模式有助於編寫能夠高效處理多個任務的軟體。接下來,我們將探討特定的效能設計模式,這些模式能夠提升應用程式的速度和資源利用率。

效能設計模式主要解決常見的瓶頸和最佳化挑戰,為開發人員提供經過驗證的方法來改善執行時間、減少記憶體使用量並實作可擴充套件性。在本章中,我們將重點介紹以下幾個主要主題:

  • 快取策略模式(Cache-Aside pattern)
  • 記憶化模式(Memoization pattern)
  • 延遲載入模式(Lazy Loading pattern)

技術需求

本章所需的技術環境與第一章相同。此外,還需要安裝以下額外的Python模組:

  • 使用以下命令安裝Faker模組:python -m pip install faker
  • 使用以下命令安裝Redis模組:python -m pip install redis
  • 使用Docker安裝並執行Redis伺服器:docker run --name myredis -p 6379:6379 redis

快取策略模式

在資料讀取頻率遠高於更新頻率的情況下,應用程式通常會使用快取來最佳化重複存取儲存在資料函式庫或資料儲存中的資訊。某些系統內建了這種快取機制,可以自動運作。如果沒有這種機制,我們需要在應用程式中自行實作適合特定使用案例的快取策略。

其中一種策略稱為快取策略模式(Cache-Aside),透過將頻繁存取的資料儲存在快取中,減少了重複從資料儲存中擷取資料的需求,從而提升效能。

實際範例

以下是軟體領域中的幾個範例:

  • Memcached是一種常用的快取伺服器。它是一種流行的記憶體內鍵值儲存,用於儲存資料函式庫呼叫、API呼叫或HTML頁面內容的結果。
  • Redis是另一種用於快取的伺服器解決方案。目前,它是我用於快取或應用程式記憶體儲存的優先選擇,因為它在這些使用案例中表現出色。
  • Amazon ElastiCache(https://aws.amazon.com/elasticache/)是一種Web服務,可以輕鬆設定、管理和擴充套件雲端中的分散式記憶體資料儲存或快取環境。

快取策略模式的使用案例

當我們需要減少應用程式中的資料函式庫負載時,快取策略模式非常有用。透過將資料儲存在快取中,可以減少傳送到資料函式庫的查詢數量。同時,由於快取資料可以更快地被檢索,因此也有助於提高應用程式的回應速度。

值得注意的是,這種模式適用於資料變更不頻繁的情況,並且資料儲存不依賴於儲存中一組條目的一致性(多個鍵)。例如,它可能適用於某些型別的檔案儲存或資料函式庫,其中鍵永遠不會被更新,偶爾會刪除資料條目,但沒有強烈的持續服務要求(直到快取被重新整理)。

實作快取策略模式

在實作快取策略模式時,涉及資料函式庫和快取,主要步驟如下:

  1. 擷取資料專案:首先檢查快取中是否存在該專案。如果存在,則直接從快取中傳回該專案。如果不存在,則從資料函式庫中讀取資料,並將其存入快取中,然後傳回。
  2. 更新資料專案:將資料專案寫入資料函式庫,並從快取中移除對應的條目。

讓我們嘗試一個簡單的實作範例,使用一個儲存名言的資料函式庫,並透過應用程式檢索名言。我們的重點將放在實作第一部分。

首先,我們需要安裝以下額外的軟體依賴:

  • SQLite資料函式庫,因為我們可以使用Python的標準模組sqlite3查詢SQLite資料函式庫。
  • Redis伺服器和redis-py Python模組。

我們將使用一個指令碼(位於ch08/cache_aside/populate_db.py檔案中)來處理資料函式庫的建立和名言表的建立,並新增範例資料。出於實用性考慮,我們還使用Faker模組在資料函式庫中產生假的名言。

我們的程式碼首先匯入所需的模組,接著進行一些常數或模組層級變數的設定:

import sqlite3
from pathlib import Path
from random import randint
import redis
from faker import Faker

fake = Faker()
DB_PATH = Path(__file__).parent / Path("quotes.sqlite3")
cache = redis.StrictRedis(host="localhost", port=6379, decode_responses=True)

圖表1:快取策略模式流程圖

圖表1翻譯:

此圖示展示了快取策略模式的基本流程。首先檢查快取中是否存在所需的資料。如果快取命中,則直接傳回快取中的資料。如果快取未命中,則從資料函式庫中讀取資料,並將其存入快取中,最後傳回資料。此流程有效地減少了對資料函式庫的直接存取次數,提升了系統的效能。

接下來,我們編寫一個函式來處理資料函式庫的設定:

def setup_db():
 try:
 cursor = db.cursor()
 cursor.execute(
 """
 CREATE TABLE quotes(id INTEGER PRIMARY KEY, text TEXT)
 """
 )

內容解密:

這段程式碼定義了一個名為setup_db的函式,用於建立名言資料函式庫表。首先,它嘗試建立一個資料函式庫連線並取得遊標。然後,使用SQL陳述式建立一個名為quotes的表,包含idtext兩個欄位。id欄位為主鍵,並自動遞增。

Memoization模式

Memoization是一種最佳化技術,主要用於加速程式的執行速度。它透過儲存耗時的函式呼叫結果,在下次遇到相同的輸入時,直接傳回快取的結果,而不需要重新計算。

Memoization的實作

在Python中,可以使用裝飾器(decorator)來實作Memoization。以下是一個簡單的範例:

def memoize(func):
 cache = dict()

 def memoized_func(*args):
 if args in cache:
 return cache[args]
 result = func(*args)
 cache[args] = result
 return result

 return memoized_func

@memoize
def fibonacci(n):
 if n < 2:
 return n
 return fibonacci(n-1) + fibonacci(n-2)

圖表2:Memoization流程圖

圖表2翻譯:

此圖示展示了Memoization技術的流程。當函式被呼叫時,首先檢查是否已經有快取的結果。如果有,則直接傳回快取的結果。如果沒有,則執行函式,計算結果,並將結果儲存到快取中,以便下次使用。這種技術有效地避免了重複計算,提升了程式的執行效率。

內容解密:

這段程式碼定義了一個名為memoize的裝飾器,用於實作Memoization技術。它使用一個字典來儲存已經計算過的結果。當函式被呼叫時,首先檢查輸入引數是否已經在快取中。如果在,則直接傳回快取的結果。如果不在,則執行函式,計算結果,並將結果儲存到快取中。透過這種方式,避免了重複計算相同的結果,大大提高了程式的執行效率。

延遲載入模式

延遲載入是一種最佳化技術,主要用於減少程式的初始載入時間和記憶體使用量。它透過延遲載入非必要的資源或資料,直到真正需要時才進行載入。

延遲載入的實作

在Python中,可以透過多種方式實作延遲載入。以下是一個簡單的範例,使用屬性(property)來實作延遲載入:

class Data:
 def __init__(self):
 self._data = None

 @property
 def data(self):
 if self._data is None:
 self._data = self.load_data()
 return self._data

 def load_data(self):
 # 模擬載入資料的過程
 print("載入資料...")
 return "資料內容"

# 使用範例
data = Data()
print(data.data) # 第一次存取時載入資料
print(data.data) # 第二次存取時直接傳回已載入的資料

圖表3:延遲載入流程圖

圖表3翻譯:

此圖示展示了延遲載入技術的流程。當存取資料時,首先檢查資料是否已經被載入。如果已經載入,則直接傳回已載入的資料。如果未載入,則進行資料載入,儲存載入的資料,並傳回。透過這種方式,避免了不必要的資料載入,減少了程式的初始載入時間和記憶體使用量。

內容解密:

這段程式碼定義了一個名為Data的類別,用於實作延遲載入技術。它使用一個私有屬性_data來儲存資料,並透過data屬性來控制資料的載入。當第一次存取data屬性時,如果資料尚未載入,則呼叫load_data方法載入資料,並將結果儲存到_data中。之後再次存取data屬性時,直接傳回已經載入的資料,避免了重複載入。

快取策略模式在效能最佳化中的應用

在軟體開發中,快取策略模式是一種重要的效能最佳化技術。本文將深入探討兩種常見的快取策略模式:快取輔助模式(Cache-Aside)與記憶化模式(Memoization),並透過具體例項與程式碼解析其運作原理與實務應用。

快取輔助模式詳解

快取輔助模式是一種常見的快取策略,其核心思想是應用程式直接控制快取與資料函式庫之間的資料一致性。以下是一個具體的實作案例:

資料函式庫初始化與資料插入

首先,我們需要建立資料函式庫並初始化資料表。以下程式碼展示瞭如何建立一個名為quotes的資料表:

import sqlite3
from pathlib import Path
import redis
from random import randint
from faker import Faker

# 資料函式庫連線設定
DB_PATH = Path(__file__).parent / Path("quotes.sqlite3")
db = sqlite3.connect(DB_PATH)

# Redis快取設定
cache = redis.StrictRedis(host="localhost", port=6379, decode_responses=True)

def setup_db():
    try:
        cursor = db.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS quotes (
                id INTEGER PRIMARY KEY,
                text TEXT NOT NULL UNIQUE
            )
        """)
        db.commit()
        print("資料表 'quotes' 建立成功")
    except Exception as e:
        print(e)

#### 程式碼解析:
此函式負責建立名為`quotes`的資料表包含id與text兩個欄位透過`CREATE TABLE IF NOT EXISTS`陳述式確保資料表只在不存在時才建立避免重複建立錯誤

資料插入功能實作

接下來,我們實作一個函式來新增多筆名言到資料函式庫中:

def add_quotes(quotes_list):
    added = []
    try:
        cursor = db.cursor()
        for quote_text in quotes_list:
            quote_id = randint(1, 100)  # nosec
            quote = (quote_id, quote_text)
            cursor.execute("""
                INSERT OR IGNORE INTO quotes(id, text)
                VALUES(?, ?)
            """, quote)
            added.append(quote)
        db.commit()
        return added
    except Exception as e:
        print(e)

#### 程式碼解析:
1. 此函式接收一個名言列表作為輸入引數
2. 對每則名言隨機產生一個id並嘗試插入資料函式庫
3. 使用`INSERT OR IGNORE`陳述式避免重複插入相同資料
4. 成功插入的資料會被記錄並傳回

主程式邏輯實作

主程式提供了三種操作模式:初始化資料函式庫、更新資料函式庫並快取、僅更新資料函式庫。以下程式碼展示了主程式的實作邏輯:

def main():
    mode = input("選擇操作模式 (init/update_db_only/update_all): ")
    if mode.lower() == "init":
        setup_db()
    elif mode.lower() in ["update_all", "update_db_only"]:
        quotes_list = [Faker().sentence() for _ in range(1, 11)]
        added = add_quotes(quotes_list)
        if added:
            print("已新增的名言:")
            for q in added:
                print(f"新增到資料函式庫:{q}")
                if mode.lower() == "update_all":
                    cache.set(str(q[0]), q[1], ex=60)
                    print("同時新增到快取")

#### 程式碼解析:
1. 程式根據使用者輸入的操作模式執行不同邏輯
2.`update_all`模式下除了更新資料函式庫還會同步更新快取
3. 使用Faker套件產生虛擬名言資料
4. 資料的有效期設定為60秒

快取讀取實作

為了示範快取輔助模式的運作,我們實作了一個獨立的程式來讀取名言資料:

def get_quote(quote_id: str) -> str:
    out = []
    quote = cache.get(f"quote.{quote_id}")
    if quote is None:
        query_fmt = "SELECT text FROM quotes WHERE id = {}"
        try:
            cursor = db.cursor()
            res = cursor.execute(query_fmt.format(quote_id)).fetchone()
            if not res:
                return "找不到對應的名言!"
            quote = res[0]
            out.append(f"從資料函式庫取得:'{quote}'")
            cache.set(f"quote.{quote_id}", quote, ex=60)
            out.append(f"已新增到快取,鍵值:'quote.{quote_id}'")
        except Exception as e:
            print(e)
    else:
        out.append(f"從快取取得:'{quote}'")
    return " - ".join(out)

#### 圖表說明:
```plantuml
@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle

title 效能設計模式提升應用程式速度

package "Python 應用架構" {
    package "應用層" {
        component [主程式] as main
        component [模組/套件] as modules
        component [設定檔] as config
    }

    package "框架層" {
        component [Web 框架] as web
        component [ORM] as orm
        component [非同步處理] as async
    }

    package "資料層" {
        database [資料庫] as db
        component [快取] as cache
        component [檔案系統] as fs
    }
}

main --> modules : 匯入模組
main --> config : 載入設定
modules --> web : HTTP 處理
web --> orm : 資料操作
orm --> db : 持久化
web --> cache : 快取查詢
web --> async : 背景任務
async --> fs : 檔案處理

note right of web
  Flask / FastAPI / Django
end note

@enduml

圖表翻譯:

此時序圖展示了快取輔助模式的工作流程。當客戶端請求資料時,系統首先查詢快取。若快取命中,則直接傳回資料;若未命中,則向資料函式庫查詢,並將結果存入快取後傳回給客戶端。

記憶化模式詳解

記憶化模式是一種透過快取函式執行結果來避免重複計算的技術。以下是一個簡單的實作範例:

def memoize(func):
    cache = dict()

    def memoized_func(*args):
        if args in cache:
            return cache[args]
        result = func(*args)
        cache[args] = result
        return result
    return memoized_func

@memoize
def expensive_computation(x):
    # 模擬耗時運算
    import time
    time.sleep(2)
    return x * x

#### 程式碼解析:
1. `memoize`函式是一個裝飾器用於快取函式執行結果
2. 當函式使用相同引數呼叫時直接傳回快取結果
3. 首次呼叫時執行實際運算並快取結果
4. 有效避免重複的耗時運算