使用適用於 PostgreSQL 的 Azure 資料庫靈活伺服器的快取提升效能

當你在 適用於 PostgreSQL 的 Azure 資料庫 靈活伺服器上建置應用程式時，新增快取層是提升回應時間、減輕資料庫負載及提升韌性最有效的方法之一。 透過從記憶體快取中提供頻繁讀取的資料，你的應用程式會減少向 PostgreSQL 發送的查詢次數。 這代表較低的 CPU 和 IOPS 消耗，因此你可以在較小的運算層級上運行，擴展讀取而不需擴大伺服器規模，並吸收流量尖峰。 快取也能增加韌性。 如果 PostgreSQL 發生短暫中斷，命中快取中既有資料的請求仍可成功處理，因此在資料庫復原期間，讀取路徑仍可維持可用。

本文將協助你判斷快取何時有效，以及哪種模式適合你的應用。 接著，它使用 Azure Managed Redis、redis 和 psycopg 程式庫，以及 Microsoft Entra ID 驗證，在 Python 中實作四種快取模式 (另行快取、參考資料預先擷取、寫入和事件驅動)。

何時新增快取

PostgreSQL 已經將經常存取的資料頁快取到緩衝區快取中，並且同時也受益於作業系統的檔案快取。 這些快取讓重複存取更快，但它們會與查詢執行及其他程序共享分配給資料庫伺服器的記憶體。 經過維護和容錯移轉作業後，內容物也需要重新預熱。

你可以透過擴展到有更多記憶體的計算選項來獲得更多資料庫快取容量。 你也可以調整 PostgreSQL 的記憶體設定，但將更多記憶體分配給緩衝快取，會使留給查詢執行和作業系統的記憶體變少。 請仔細測試記憶體變更，以避免發生記憶體不足的情況。

Azure Managed Redis 則補充這些原生快取。 它將選定的應用程式資料與查詢結果儲存在資料庫伺服器之外，提供較低延遲的存取權限，適用於時間敏感的讀取路徑，並能在 PostgreSQL 恢復或預熱快取時保持快取讀取的可用性。 當這些好處足以證明增加額外的應用程式邏輯以及再營運一個服務是合理的時候，再使用它。 它並不能取代 PostgreSQL 作為真實來源。

當您的工作負載具備以下特性時，新增像 Azure Managed Redis 這樣的外部快取最有幫助：

• 以讀取為主的存取模式。 相同的列被讀取的頻率遠高於它們的變更，例如產品目錄、使用者資料、組態資料或參考表。
• 高成本或重複的查詢。 聚合、聯結或計算結果，其生成成本高，但在短時間窗口內保持穩定。
• 延遲敏感的端點。 面向使用者的作業，記憶體內讀取 (亞毫秒) 比資料庫往返更為可取。
• 可預測的尖峰。 季節性或事件驅動型流量，其中快取可以吸收負載，否則將迫使您擴展計算能力。
• 維護與故障切換敏感度。 在 PostgreSQL 執行個體進行維護或容錯移轉作業後復原或預熱快取時，需要穩定回應時間的讀取路徑。

對於寫入密集的工作負載、必須始終維持交易一致性的資料，或本來就很快且很少重複執行的查詢，快取的效果較有限。

快取模式

本文以零售店面作為貫穿全文的範例。 應用程式的不同部分適合採用不同的快取模式。 以下章節將實作 Python 的前四個模式。 文章描述了會話卸載與狀態卸載以及多區域快取模式，但未提供這些程式碼實作。

先決條件

• 一個有有效訂閱的 Azure 帳號。 免費註冊帳號。
• 一個啟用 Microsoft Entra 認證的 適用於 PostgreSQL 的 Azure 資料庫 靈活伺服器實例。 要建立一個，請參見「Create an 適用於 PostgreSQL 的 Azure 資料庫 flexible server」。
• 一個 Azure Managed Redis 實例。 要建立一個，請參見「建立 Azure Managed Redis 實例」。 在與 PostgreSQL 伺服器相同的區域建立（生產環境則共用同一虛擬網路），以減少延遲。
• 在兩個服務上都可以存取您的開發人員或應用程式身分：在 PostgreSQL 伺服器上有 Microsoft Entra 管理員或角色，快取則有 Redis 存取原則指派。 請參閱 適用於 適用於 PostgreSQL 的 Azure 資料庫 的 Microsoft Entra 驗證及 搭配 Azure Managed Redis 使用 Microsoft Entra ID 進行驗證。
• Python 3.10 或更新版本。
• The Azure CLI. 要安裝它，請參見如何安裝 Azure CLI。

步驟 1：安裝用戶端函式庫

安裝 Redis 和 PostgreSQL 用戶端函式庫，以及 Azure 身份函式庫以支援 Microsoft Entra 認證。

pip install "redis>=5.0,<6.0" "psycopg[binary]>=3.1,<4.0" "azure-identity>=1.17,<2.0"

步驟 2：連接 Microsoft Entra ID

使用 Microsoft Entra ID 認證代替存取金鑰。 Microsoft Entra ID 省去了在應用程式中儲存秘密的需求，讓你能集中管理存取權限。

以下程式碼建立 Redis 用戶端與 PostgreSQL 連線，兩者皆透過管理身份或開發者憑證 DefaultAzureCredential進行驗證。 範例中使用叢集感知 RedisCluster 用戶端，與本範例所使用的 OSS 叢集政策相符。 如果你的快取使用企業叢集政策，請改用標準 redis.Redis 用戶端。

import os
import json
import redis
from redis.cluster import RedisCluster
import psycopg
from azure.identity import DefaultAzureCredential

REDIS_HOST = os.environ["REDIS_HOST"]  # for example, mycache.eastus.redis.azure.net
REDIS_PORT = 10000
PG_HOST = os.environ["PG_HOST"]        # for example, myserver.postgres.database.azure.com
PG_DATABASE = os.environ["PG_DATABASE"]

credential = DefaultAzureCredential()

# Acquire a token for Azure Managed Redis and use it as the password.
redis_token = credential.get_token("https://redis.azure.com/.default")

# The username is the object ID of the Microsoft Entra identity.
redis_client = RedisCluster(
    host=REDIS_HOST,
    port=REDIS_PORT,
    ssl=True,
    ssl_check_hostname=False,  # cluster nodes are reached by IP; the certificate chain is still validated
    username=os.environ["REDIS_USER_OBJECT_ID"],
    password=redis_token.token,
    decode_responses=True,
)

# Acquire a token for Azure Database for PostgreSQL and use it as the password.
pg_token = credential.get_token("https://ossrdbms-aad.database.windows.net/.default")

pg_conn = psycopg.connect(
    host=PG_HOST,
    dbname=PG_DATABASE,
    user=os.environ["PG_USER"],
    password=pg_token.token,
    sslmode="require",
)

步驟 3：另行快取

另行快取是最常見的模式，並在此範例中用於讀取產品資料。 應用程式會先檢查 Redis，若未命中則查詢 PostgreSQL，並將結果寫入快取，同時設定存活時間（TTL）。 少數熱門產品會帶來最多閱讀量，因此點擊率很高。

由於大多數讀取是從記憶體中執行，另行快取會持續減輕 PostgreSQL 的讀取負載。 這種減少意味著連線數量減少、緩衝快取流失減少，以及降低 CPU 和 IOPS。 您可以在不升級伺服器規格或增加讀取複本的情況下，因應讀取流量尖峰。 您只有在快取未命中時 (第一次存取，或 TTL 到期後) 才會查詢 PostgreSQL。 請參閱 「尋找需要快取的內容 」以辨識值得快取的查詢。

CACHE_TTL_SECONDS = 300  # 5 minutes

def get_product(product_id: int) -> dict | None:
    cache_key = f"product:{product_id}"

    # 1. Try the cache first.
    cached = redis_client.get(cache_key)
    if cached is not None:
        return json.loads(cached)

    # 2. On a miss, read from PostgreSQL.
    with pg_conn.cursor() as cursor:
        cursor.execute("SELECT id, name, price FROM products WHERE id = %s", (product_id,))
        row = cursor.fetchone()
    if row is None:
        return None

    product = {"id": row[0], "name": row[1], "price": float(row[2])}

    # 3. Populate the cache with a TTL, then return.
    redis_client.set(cache_key, json.dumps(product), ex=CACHE_TTL_SECONDS)
    return product

步驟 4：參考資料預取

穩定且經常被讀取、但很少變更的資料 (例如分類、品牌或配送設定) 不必等到快取未命中時才處理。 預先將其載入快取，並在來源變更時重新整理。 在 PostgreSQL 架構中，這些通常是小型的查找表和維度表，會被合併成許多查詢。 直接從記憶體提供這些資料，可免除資料庫中大量重複的聯結和查找作業。 與另行快取不同，這種方式不會發生每次請求都可能出現的未命中問題，也沒有 TTL 競爭。 每次更新都會重新整理頁面，所以讀取總是暖身狀態。

def prefetch_categories() -> None:
    with pg_conn.cursor() as cursor:
        cursor.execute("SELECT id, name FROM categories ORDER BY name")
        categories = [{"id": r[0], "name": r[1]} for r in cursor.fetchall()]
    redis_client.set("ref:categories", json.dumps(categories))  # no TTL; refreshed on change

def get_categories() -> list[dict]:
    cached = redis_client.get("ref:categories")
    return json.loads(cached) if cached else []

步驟 5：寫入

當變更必須立即可見時，應在同一操作中寫入 PostgreSQL 與快取，而非等待 TTL 到期或使金鑰失效。 例如，此模式用於更新樣本中的價格資訊。 PostgreSQL 依然是真相的來源。 變更會先在該處提交，然後更新快取，因此在寫入之後進行讀取會回傳新值。

當寫入兩個未共用同一交易的系統時，資料庫提交可能會成功，但快取更新卻失敗，而且沒有簡單的解法。 以下程式碼片段展示了順利執行的情境，並省略失敗處理。在生產環境中，你需要決定如何處理刷新失敗，例如當失敗看起來只是暫時性問題時進行重試，或使該金鑰失效，讓下一次讀取時從 PostgreSQL 重新載入。 無論哪種方式，PostgreSQL 都保持正確的值，因此陳舊或缺失的快取條目總是可以恢復的。 當快取更新必須可靠落地時，應將其從資料庫的變更串流中移除 (參見事件驅動失效)。

def update_price(product_id: int, new_price: float) -> None:
    # 1. Write to PostgreSQL, the source of truth.
    with pg_conn.cursor() as cursor:
        cursor.execute("UPDATE products SET price = %s WHERE id = %s", (new_price, product_id))
    pg_conn.commit()

    # 2. Refresh the cached entry so reads see the new price right away.
    with pg_conn.cursor() as cursor:
        cursor.execute("SELECT id, name, price FROM products WHERE id = %s", (product_id,))
        row = cursor.fetchone()
    if row is not None:
        product = {"id": row[0], "name": row[1], "price": float(row[2])}
        redis_client.set(f"product:{product_id}", json.dumps(product), ex=CACHE_TTL_SECONDS)

步驟 6：事件驅動失效

事件驅動的失效機制藉由回應資料變更事件，使快取與資料庫保持一致。 它會隨著項目變動而更新或失效，而不是在計時器上過期。 寫入者會在耐用的 Redis 串流中附加事件，並建立僅附加的日誌，然後一位或多位使用者讀取這些事件並更新快取。 由於串流會持續存在，因此即使消費者重新啟動，事件仍會保留下來。 消費者團體會將每個事件交付給單一工作者，追蹤確認，確保不會遺失或重複處理，並允許跨工作者擴展處理。

當快取值從其他地方變動的資料 (狀態、投影或聚合) 衍生而來時，請使用此模式，而 TTL 則可能提供過時資料或強制不斷重算。 在實體店中，這種模式透過訂單履行來驅動訂單狀態的更新。 下單會將訂單寫入 PostgreSQL，快取其初始狀態，並在串流中附加事件 placed ：

ORDER_STREAM = "orders:events"

def place_order(product_id: int, quantity: int) -> int:
    with pg_conn.cursor() as cursor:
        cursor.execute(
            "INSERT INTO orders (product_id, quantity, status) VALUES (%s, %s, 'placed') RETURNING id",
            (product_id, quantity),
        )
        order_id = cursor.fetchone()[0]
    pg_conn.commit()

    redis_client.set(f"order:{order_id}:status", "placed", ex=86400)
    redis_client.xadd(ORDER_STREAM, {"order_id": order_id, "status": "placed"}, maxlen=10000, approximate=True)
    return order_id

訂單履行工作程序會在消費者群組中運作：它會讀取新事件、推進 PostgreSQL 中各筆訂單的處理進度、更新快取中的 order:{id}:status 投影，並確認該事件。 訂單頁面會讀取這個預測，所以狀態檢查會快速且不會碰到資料庫。 這個數值會保持正確，是因為事件讓它保持最新。

GROUP = "fulfillment"

def process_orders() -> None:
    try:
        redis_client.xgroup_create(ORDER_STREAM, GROUP, id="0", mkstream=True)
    except redis.exceptions.ResponseError:
        pass  # group already exists

    while True:
        events = redis_client.xreadgroup(GROUP, "worker-1", {ORDER_STREAM: ">"}, count=10, block=5000)
        for _stream, entries in events or []:
            for event_id, fields in entries:
                order_id = int(fields["order_id"])
                with pg_conn.cursor() as cursor:
                    cursor.execute("UPDATE orders SET status = 'shipped' WHERE id = %s", (order_id,))
                pg_conn.commit()
                redis_client.set(f"order:{order_id}:status", "shipped", ex=86400)
                redis_client.xack(ORDER_STREAM, GROUP, event_id)

本範例中的事件來源是應用程式，它撰寫 PostgreSQL，並將事件附加於相同路徑。 PostgreSQL 也能自行發送變更： LISTEN/NOTIFY 用於輕量級通知，或邏輯解碼（變更資料擷取）以建立持久的列級變更串流。 讓快取由 PostgreSQL 自身的變更串流驅動，意味著它會對每一筆已提交的變更做出反應，甚至包括繞過應用程式進行的寫入。

最佳實務

遵循這些做法，可以確保您的快取正確、高效且具成本效益。

• 在任何有資料過時風險的地方都應設定 TTL。 TTL 在失效失敗時限制了過時性。 將 TTL 與應用程式能接受的最大過時度相匹配。 只有在有可靠的無效化與刷新流程時，才使用長 TTL 或不使用 TTL 作為參考資料。
• 使用一致的金鑰命名方式。 命名空間金鑰會依服務、實體和識別碼命名，例如 product:42 或 user:1001:profile。 當鍵格式或值結構可能變更時，新增版本。
• 快取適當的粒度。 當個別實體或小型結果集經常重複使用，且容易使其失效時，請快取它們。 不要快取重複使用率低的資料。 過度快取會浪費記憶體，並降低命中率。
• 妥善處理快取未命中和服務中斷。 將快取視為一種最佳化機制，而非權威資料來源。 如果 Redis 無法使用，請改用 PostgreSQL 作為受限的備援方案。 新增逾時、斷路器、退後及請求限制，以保護 PostgreSQL。 如果 PostgreSQL 有短暫中斷，你可以繼續為快取中已有的資料提供讀取，而寫入則等待資料庫恢復。
• 防止快取雪崩。 當熱門金鑰過期時，資料庫可能會同時收到多個請求。 使用 TTL 抖動、請求合併、過期時重新驗證或短分散式鎖定，讓一個請求重新填入項目。
• 將快取調整為適當大小。 監控命中率、記憶體使用率、淘汰率、過期率、延遲、快捷鍵和鍵基數。 低命中率可能表示快取容量過小、過度淘汰、金鑰選取不佳，或存取模式不佳。 關於規模調整指引，請參閱 Azure Managed Redis 的層級選擇指引。
• 選擇適合您金鑰的驅逐原則。 對於純快取資料庫，請先使用 allkeys-lru 或 allkeys-lfu。 只有當同一資料庫包含過期快取金鑰和受保護的非過期金鑰時，才使用揮發性-* 策略。 當沒有金鑰具有 TTL 時，不穩定原則可能會停止驅逐操作。 盡可能將快取資料與受保護的資料分開。
• 有效率地序列化。 JSON 是可讀且可攜的。 對於高吞吐量路徑，建議測試緊湊的二進位格式以減少記憶體與網路負擔。 在更換格式前，先評估記憶體、CPU、延遲、結構演進和除錯對效能的影響。

找出要快取的項目

最有效的快取目標是應用程式最常執行的查詢，針對變動最小的資料。 與其猜測哪些查詢符合此描述，不如比較 適用於 PostgreSQL 的 Azure 資料庫 啟用相關功能後，查詢遙測中能收集的歷史與當前視圖：

• 查詢存放區 會持續保存查詢執行統計資料以進行歷史分析。 利用呼叫次數以及總執行時間和平均執行時間，找出在較長時間內持續主導資料庫負載的查詢。 請參見使用 查詢存放區 監控效能。
• Query Performance Insight 在 Azure 入口網站中視覺化 查詢存放區 資料，讓您能發現頻繁且資源密集的查詢，並比較其隨時間的行為。 請參見 查詢效能洞察。
• pg_stat_statements 在資料庫中公開每語句累積統計資料，直接檢視目前觀察視窗。 它的統計數據可以重置，所以當你需要在觀察視窗間保留歷史時，請使用 查詢存放區。

在選擇快取候選前，請同時使用兩種視角。 短期的峰值可能不代表工作負載的正常行為，而歷史平均值則可能掩蓋當前的迴歸。 優先處理頻繁 、昂貴且穩定的查詢，意即呼叫次數多、總執行時間長，且每個請求結果不會改變。 這些查詢帶來最高的快取命中率和最大的資料庫負載下降。 一個持續執行但每分鐘回傳相同列的查詢，例如產品列表、分類樹或定價表，是理想的候選對象。

相關內容

• Azure Managed Redis 是什麼？
• 快取指引
• 規劃適用於 PostgreSQL 的 Azure Database 彈性伺服器部署，以提升作業效能