iT邦幫忙

2025 iThome 鐵人賽

DAY 18
0

延續前一篇的效能觀測(cProfile / py-spy / line-profiler),我們今天把「量測之後的行動」落到快取層,讓系統用更少的資源交付更快的回應。這正符合系列路線圖在 Day 18 預告的主題安排。


為什麼要快取?先量測再下刀

快取不是萬靈藥。請先用 Day 17 的方法找出瓶頸,再評估是否屬於這兩類情境:

  • CPU 密集:重複計算、昂貴的序列化或模板渲染。適合做函式層級快取。
  • I/O 密集:外部 API、資料庫查詢、檔案系統。適合做跨程序或跨機快取(例如 Redis)。

分層觀點:從函式到分散式

  1. 應用內(In-Process)functools.lru_cache 省去相同輸入的重算,零依賴、延遲最低,但僅限單程序且記憶體受限。
  2. 程序間(Cross-Process):Redis 讓多實例共享結果,能設定 TTL 與失效策略,適合讀多寫少的熱資料。
  3. 邏輯配套:失效設計、Stampede 防護、負向快取、Key 命名、序列化格式(建議用 Day 15 的 orjson)。

lru_cache:零依賴的函式級加速

基本用法

from functools import lru_cache

@lru_cache(maxsize=1024)  # 最近最常用 1024 個結果
def heavy_calc(a: int, b: int) -> int:
    # 昂貴計算...
    return a * a + b * b

  • maxsize 決定記憶體占用與命中率的取捨。
  • typed=True 可分別快取 11.0 這類值。

失效與監控

heavy_calc.cache_info()     # 命中/未命中統計
heavy_calc.cache_clear()    # 全清(部署切版或資料規模變更時)

什麼時候不該用

  • 需要跨實例共享結果。
  • 回傳值會依「隱性狀態」改變(例如讀環境變數、時鐘)。這會讓快取鍵不穩定,請把影響輸入顯式化。

提醒:多程序(gunicorn workers)或多 Pod 情境下,應用內快取彼此不共享。要共享請上 Redis。


Redis:共享快取與 TTL 策略

連線與序列化

import orjson
import redis                         # redis-py
from typing import Any

r = redis.Redis.from_url("redis://localhost:6379/0", socket_timeout=1.0)

def dumps(x: Any) -> bytes:
    return orjson.dumps(x)

def loads(b: bytes) -> Any:
    return orjson.loads(b)

  • orjson 加速序列化/反序列化,降低 CPU 負擔。
  • 連線請設定 timeout 與連線池,避免資源被吃滿。

Cache-Aside(最常見)

def get_user(uid: str):
    key = f"v1:user:{uid}"
    cached = r.get(key)
    if cached is not None:
        return loads(cached)

    data = db_fetch_user(uid)              # 真實資料來源
    if data is None:
        r.setex(key, 60, dumps({"__nil": True}))  # 負向快取 60s,擋住重複查無
        return None

    r.setex(key, 300, dumps(data))         # 設定 TTL 5 分鐘
    return data

  • 負向快取:不存在的結果也短暫快取,緩解「熱不存在鍵」風暴。
  • Key 命名{版本}:{領域}:{識別},有助於日後全域升版(見下)。

失效設計(Invalidation)

  • 主寫入路徑清快取:資料變動後,刪除或更新相對應鍵。
  • 版本前綴:將 key 前綴帶版本號,例如 v1:。當資料模型或序列化變更時,只要把前端程式升到 v2:,舊鍵自然過期,不必遍歷刪除。
  • 同步難題:寫入與刪除可能競爭。若能接受短暫舊資料,採「先寫 DB 再刪 cache」;若一致性要求更高,考慮「寫 DB 成功後發佈 Pub/Sub 事件,各服務自行失效對應鍵」。

Stampede(羊群效應)防護

大量請求同時遇到快取失效會一起打到來源,導致雪崩。常見手法:

  • Jitter TTL:在 TTL 上加 0~N 秒隨機抖動,避免集體同時失效。
  • Soft TTL + 續租:在資料結構中放 expires_at_softexpires_at_hard。軟到期時允許舊值回應,並由第一個進入的請求用短鎖進行重建。
  • 分布式鎖(短暫):用 SET lock:key val NX EX 5 搭配極短過期,只有鎖持有者重建快取,其餘回舊值或稍候重試。

寫入策略選擇

  • Cache-Aside:最普遍,讀取時查 cache,不在 cache 時回源並回填。寫入只動 DB,再刪 cache。
  • Write-Through:所有寫入同時更新 DB 與 cache。寫延遲略高,讀一致性更好。
  • Write-Back:先寫 cache,延後批次寫 DB。延遲最低,但故障風險高,需嚴謹併發與持久化策略。

失敗即常態:配上重試與降級

外部快取或資料來源都會出錯,請沿用 Day 14 的策略:可恢復錯誤重試,不可恢復則降級回舊值或預設值,並寫結構化日誌。

import structlog
from tenacity import retry, stop_after_attempt, wait_exponential, RetryError

log = structlog.get_logger()

@retry(stop=stop_after_attempt(3), wait=wait_exponential(1, 2, 8))
def rebuild_user(uid: str):
    data = db_fetch_user(uid)           # 可能 timeout
    r.setex(f"v1:user:{uid}", 300, dumps(data))
    return data

def get_user_safe(uid: str):
    key = f"v1:user:{uid}"
    if (b := r.get(key)) is not None:
        return loads(b)

    try:
        return rebuild_user(uid)
    except RetryError as e:
        log.warning("cache_rebuild_failed", uid=uid, reason=str(e))
        return {"status": "degraded"}   # 降級回應

日誌請使用 Day 13 的 JSON 結構化做法,保留 uid、cache_key、hit/miss、latency_ms 等欄位,方便在觀測平台聚合分析。


非同步情境與多實例

  • Async:在 FastAPI 或 anyio 場景,使用 redis.asyncio 並避免在協程中呼叫同步 I/O。測試時用 pytest-asyncio
  • 多實例:單機 lru_cache 命中率會被分散;跨實例共享必須靠 Redis。

明天我們會往資料層前進,將快取策略與 ORM/查詢邏輯一併納入工程化設計(預告見 Day 1 路線圖)。


上一篇
Day 17 -效能觀測:cProfile、py-spy、line-profiler
下一篇
Day 19 -資料層工程化:SQLAlchemy 2.x 與 Repository Pattern
系列文
30 天 Python 專案工坊:環境、結構、測試到部署全打通23
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言