延續前一篇的效能觀測(cProfile / py-spy / line-profiler),我們今天把「量測之後的行動」落到快取層,讓系統用更少的資源交付更快的回應。這正符合系列路線圖在 Day 18 預告的主題安排。
快取不是萬靈藥。請先用 Day 17 的方法找出瓶頸,再評估是否屬於這兩類情境:
functools.lru_cache
省去相同輸入的重算,零依賴、延遲最低,但僅限單程序且記憶體受限。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
可分別快取 1
與 1.0
這類值。heavy_calc.cache_info() # 命中/未命中統計
heavy_calc.cache_clear() # 全清(部署切版或資料規模變更時)
提醒:多程序(gunicorn workers)或多 Pod 情境下,應用內快取彼此不共享。要共享請上 Redis。
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 負擔。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
{版本}:{領域}:{識別}
,有助於日後全域升版(見下)。v1:
。當資料模型或序列化變更時,只要把前端程式升到 v2:
,舊鍵自然過期,不必遍歷刪除。大量請求同時遇到快取失效會一起打到來源,導致雪崩。常見手法:
expires_at_soft
與 expires_at_hard
。軟到期時允許舊值回應,並由第一個進入的請求用短鎖進行重建。SET lock:key val NX EX 5
搭配極短過期,只有鎖持有者重建快取,其餘回舊值或稍候重試。外部快取或資料來源都會出錯,請沿用 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 等欄位,方便在觀測平台聚合分析。
redis.asyncio
並避免在協程中呼叫同步 I/O。測試時用 pytest-asyncio
。lru_cache
命中率會被分散;跨實例共享必須靠 Redis。明天我們會往資料層前進,將快取策略與 ORM/查詢邏輯一併納入工程化設計(預告見 Day 1 路線圖)。