Web Day 20 快取:Redis 與快取策略
執行需求:CPU 可跑;Redis 容器需 Docker(無 Redis 時可用 fakeredis 或記憶體 LRU 快取)。今天是「Web 系統實戰:用 FastAPI 打造能上線的後端」系列的第二十篇。昨天用 Redis 8 當 Celery 的 broker,今天把它換成「快取」:把同一筆資料暫存在記憶體裡,讓第二次以後的存取直接命中。今天會把 cache-aside、TTL、stampede 一次講清楚,並給出完整的可執行程式碼與測試。
引言
「快取」聽起來很直觀,但實作起來坑很多:什麼時候該寫快取、寫進去的資料怎麼更新、過期時間怎麼設、快取不在的時候怎麼辦、快取被穿透怎麼辦。這些問題在小型服務裡或許感覺不到,但流量一上來就會變成 P0 等級的故障。今天的做法是先講觀念、再寫程式碼、最後用一個整合測試驗證快取真的有效。
今天的範例同樣接續 Day 16-19 的專案。我們會做四件事:第一,解釋 cache-aside 與其他常見策略;第二,用 redis-py 5.x 連線並寫一個 cache-aside 函式;第三,用 FastAPI Depends 注入 Redis 連線,並示範「無 Redis 也能用」的兩種替代方案(fakeredis、lru_cache);第四,整理常見踩雷並做效能比較。
快取的觀念:為什麼要、為什麼不
快取的核心想法是「讀多寫少」。當一個查詢每天被呼叫一萬次但每天只更新兩次,你花資源重算一萬次是浪費;把結果暫存起來、第二次以後直接讀,就是快取。能這樣做的前提是「資料可以暫時不一致」:你接受「使用者看到的是 30 秒前的資料」。如果你的業務不能容忍延遲(例如即時股價、付款狀態),快取就不適合。
常見的快取策略有四種:
- cache-aside(最常見):應用程式自己管快取。讀取時先看快取,沒有就讀資料庫、把結果寫進快取;寫入時先更新資料庫,再讓快取過期(不是同步更新快取)。
- read-through:快取層自己負責去讀資料庫。應用程式只看快取;快取層負責「miss 時去抓」。
- write-through:寫入時同步寫進快取與資料庫。讀寫都先經過快取層。
- write-behind:寫入時只寫快取,快取層非同步寫進資料庫。效能最好但有掉資料風險。
Web 後端最常用的是 cache-aside,因為它簡單、明示把快取邏輯放在應用層,並且「過期」比「同步」更可靠。Web Day 35-44 的預約管理系統也會用 cache-aside。今天的範例就圍繞這個策略展開。
Redis 8 與 redis-py 5.x:基本操作
Redis 8 是 2025 年 4 月發布的版本,重點強化了向量搜尋、JSON 結構、ACL 等特性,但對純快取場景來說,新舊版的差異不大。我們用 redis-py 5.x 連線:
# app/cache.py
# 快取模組:集中管理 Redis 連線與 cache-aside 函式
import json
from typing import Any, Callable
import redis
# 在測試環境可以換成 fakeredis
redis_client: redis.Redis = redis.Redis.from_url(
"redis://localhost:6379/2",
decode_responses=True,
)
def get_redis() -> redis.Redis:
# 用 Depends 注入的版本
return redis_client
decode_responses=True 讓回傳的字串自動解碼成 Python 字串,否則會是 bytes。網址 redis://localhost:6379/2 的 /2 是邏輯資料庫編號,跟 Celery 用的 /0、/1 隔開,避免互相干擾。
基本的 cache-aside 函式長這樣:
# app/cache.py(接續)
def cache_aside(
key: str,
loader: Callable[[], Any],
ttl_seconds: int = 60,
) -> Any:
"""cache-aside 讀取:先看快取,沒有就呼叫 loader 並把結果寫進快取。"""
cached = redis_client.get(key)
if cached is not None:
return json.loads(cached)
value = loader()
redis_client.set(key, json.dumps(value), ex=ttl_seconds)
return value
def invalidate(key: str) -> None:
"""寫入或更新時主動讓快取過期。"""
redis_client.delete(key)
cache_aside 把「讀快取 → 沒有就 loader → 寫回快取」這套流程封裝起來;invalidate 則是「資料變更時呼叫,讓快取失效」。這兩個函式就是 cache-aside 策略的核心 API。
序列化用 JSON 是因為它跨語言、debug 時可直接讀;但 JSON 不能直接序列化 datetime、Decimal、Pydantic 模型。我們會在端點裡先把資料轉成 dict 再快取:
# app/main.py
# 用 cache_aside 包裝一個實際的端點
from datetime import datetime
from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from sqlmodel import select
from app.cache import cache_aside, get_redis, invalidate
from app.db import get_db
from app.models import Item
app = FastAPI(title="Web Day 20 範例", version="0.20.0")
@app.get("/items/{item_id}")
def get_item(item_id: int, db: Session = Depends(get_db)):
def loader():
item = db.get(Item, item_id)
if item is None:
return None
return {
"id": item.id,
"name": item.name,
"price": item.price,
"fetched_at": datetime.utcnow().isoformat(),
}
return cache_aside(f"item:{item_id}", loader, ttl_seconds=30)
@app.put("/items/{item_id}/price")
def update_price(item_id: int, new_price: int, db: Session = Depends(get_db)):
item = db.get(Item, item_id)
if item is None:
return {"error": "not found"}
item.price = new_price
db.add(item)
db.commit()
invalidate(f"item:{item_id}")
return {"status": "updated"}
GET 端點用 cache_aside 把讀取包起來,TTL 設 30 秒;PUT 端點改完資料後主動呼叫 invalidate,讓下一個 GET 重新從資料庫撈。這個寫法的好處是「寫入簡單、讀取自動命中」。
完整實作:可測試的快取層
要測試 cache-aside,關鍵是「能在測試裡換掉 Redis」。最常見的寫法是用 fakeredis:它是一個純 Python 的 Redis 模擬器,介面與 redis-py 幾乎一致,但完全跑在記憶體裡。
# 命令列:安裝 fakeredis(測試用)
uv add --dev "fakeredis==2.26"
# tests/conftest.py(加上 fakeredis fixture)
import fakeredis
import pytest
import app.cache
@pytest.fixture
def fake_redis(monkeypatch):
# 建立一個記憶體的 Redis 模擬器
client = fakeredis.FakeRedis(decode_responses=True)
# 換掉 app.cache 模組的全域 redis_client
monkeypatch.setattr(app.cache, "redis_client", client)
yield client
client.flushall()
這個 fixture 把 app.cache.redis_client 換成 fakeredis 物件,所有 cache_aside 與 invalidate 呼叫都會走這個記憶體版本。flushall 在 teardown 階段確保下一個測試從空白開始。
接著寫 cache-aside 的整合測試:
# tests/test_cache.py
# 驗證 cache-aside 的讀取路徑與失效路徑
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_get_item_uses_cache_after_first_call(client, fake_redis):
# 第一次:從資料庫抓
response = client.get("/items/1")
assert response.status_code == 200
body1 = response.json()
# 第二次:應該從快取抓;fake_redis 已經有資料
response = client.get("/items/1")
assert response.status_code == 200
body2 = response.json()
assert body2 == body1
def test_invalidate_removes_cached_item(client, fake_redis):
client.get("/items/1")
assert fake_redis.exists("item:1") == 1
# 寫入操作讓快取失效
response = client.put("/items/1/price", params={"new_price": 999})
assert response.status_code == 200
assert fake_redis.exists("item:1") == 0
def test_ttl_expires_cached_value(client, fake_redis, monkeypatch):
# 縮短 TTL:把 cache_aside 的預設值改成 0 秒
import app.cache
monkeypatch.setattr(app.cache, "cache_aside",
lambda key, loader, ttl_seconds=0: loader())
client.get("/items/1")
# 因為 TTL=0,寫入後立刻過期
assert fake_redis.exists("item:1") == 0
這三個測試合在一起驗證了 cache-aside 的三個關鍵特性:第一次讀會從資料庫撈、第二次讀會命中快取、主動 invalidate 會讓快取失效。fake_redis.exists 直接讀 fakeredis 的內部狀態,確認「快取裡真的有資料」,這比只看端點回應更嚴謹。
沒有 Redis 也能用:lru_cache 與 fakeredis
在 Windows 與小型開發環境,啟動 Redis 容器有時候不方便。今天示範兩種完全不需要 Redis 的替代方案。
第一種是用 Python 標準庫 functools.lru_cache:
# app/cache.py(lru_cache 版本)
from functools import lru_cache
from typing import Any, Callable
@lru_cache(maxsize=128)
def cached_loader(item_id: int) -> dict | None:
# 注意:lru_cache 對參數有限制,只接受可雜湊的物件
from app.db import get_engine
from app.models import Item
# 這裡只能「同步」抓取,且無法用 Session
engine = get_engine()
with engine.connect() as conn:
row = conn.execute(
__import__("sqlmodel").text(
"SELECT id, name, price FROM item WHERE id = :id"
),
{"id": item_id},
).first()
if row is None:
return None
return {"id": row.id, "name": row.name, "price": row.price}
lru_cache 把函式呼叫結果暫存在記憶體裡,第二次呼叫直接命中。優點是零基礎設施;缺點是「每個行程一份快取」,多 worker 部署時快取不會共享,而且無法設 TTL(除非用 cache_clear() 手動清)。
第二種是用 fakeredis 當正式環境的後備方案:
# app/cache.py(自動切換)
import os
import fakeredis
import redis
if os.getenv("APP_ENV") == "test":
redis_client = fakeredis.FakeRedis(decode_responses=True)
else:
redis_client = redis.Redis.from_url(
os.getenv("REDIS_URL", "redis://localhost:6379/2"),
decode_responses=True,
)
這個寫法讓「測試環境」與「本機開發」自動用 fakeredis,「正式環境」才用真的 Redis。記得用環境變數切換,不要在程式碼裡寫死。
常見錯誤與踩雷
第一個雷是「把不可序列化的物件寫進快取」。今天範例裡 json.dumps(datetime.now()) 會直接拋 TypeError: Object of type datetime is not JSON serializable。解法是把 datetime 轉成 ISO 字串、把 Decimal 轉成 float、把 Pydantic 模型呼叫 .model_dump()。實務上最安全的做法是「端點輸出用 Pydantic 物件、cache_aside 的 loader 只回傳 dict」。
第二個是「快取擊穿(stampede)」。熱門資料過期的瞬間,會有大量請求同時打到資料庫,因為它們都發現快取是空的。解法有兩個方向:第一是「TTL 加隨機抖動」,讓不同 key 的過期時間不完全一致;第二是「single-flight」,用 lock 確保同一個 key 只有一個請求去抓資料庫,其他請求等待結果。Python 的 asyncio.Lock 與 Redis 的 SETNX 都能做到,今天先記得有這個觀念,後續效能篇章會再深入。
第三個是「忘記設 TTL」。沒有 TTL 的快取會一直長大,最後把 Redis 記憶體吃光。設定 ttl_seconds 是 cache-aside 的紀律之一,沒有例外。
第四個是「快取不一致」。當資料庫更新成功但快取 invalidate 失敗(例如 Redis 連線斷掉),下一個讀取會拿到舊資料。實務上你會把 invalidate 包進「try/except + 記 log」,並且接受「最差情況下,使用者看到的是過期資料」。要 100% 強一致,唯一的方法是不用快取。
cache-aside 之外的選擇:為什麼通常還是 cache-aside
為什麼不直接用 write-through 或 read-through?因為 cache-aside 的失效與否由應用程式決定,當你說「資料庫改完、讓快取過期」這是顯式的、可除錯的;write-through 雖然一致性較好,但要嘛你自己實作兩邊寫入、要嘛交給 framework,後者通常犧牲了對「什麼資料進快取」的掌控權。
write-behind 雖然效能最好,但「fastapi 行程死掉、待寫資料就掉」是個常見的雷區。對金融、訂單這類不能掉資料的場景,write-behind 不適合;對「統計瀏覽數、計數器、推薦分數」這類「偶爾掉幾筆沒關係」的場景,write-behind 是合理的選擇。實務上 80% 的快取場景用 cache-aside 就夠,剩下 20% 才需要考慮其他策略。
另外一個常被問的問題是「要 cache-aside 還是 HTTP cache(Cache-Control、ETag)?答案是兩者不互斥:HTTP cache 在「瀏覽器與 CDN」層級處理;Redis cache-aside 在「應用程式與資料庫」層級處理。一個典型的後端會同時用兩者:Redis cache-aside 處理「重複的 API 呼叫」、HTTP cache 處理「同一個使用者重複的 GET 請求」。
效能與實務提醒
第一個提醒是「用 batch 操作代替迴圈」。當你需要一次抓 100 個 key 時,MGET 比 100 次 GET 快非常多。Redis 的 pipeline 也能把多個指令合成一次網路往返。今天的範例只示範單一 key,但實務上幾乎所有「依 ID 抓資料」的場景都會用 MGET。
第二個是「監控命中率」。一個命中率 50% 的快取等同於沒裝。實務上你會想要統計 hit / miss 的比例,這可以用 redis-py 的 INFO 指令或自己在 cache_aside 裡加計數器。FastAPI 的 middleware 或獨立的 metrics endpoint 都能做到(Web Day 34 會展開監控主題)。
第三個是「快取 key 的命名」。我們用 item:{id} 這種格式,原因是容易看出「這個 key 是哪個資源的」。當你的服務有十幾種資源時,把 key 統一前綴(例如 web_day_20:item:1)能讓你在 Redis 裡一眼看出每個 key 屬於哪個應用、哪個版本,這對「Redis 容量滿了時要清誰」非常重要。
小結
今天把快取從觀念推到實作。cache-aside 是最常見的策略,Redis 8 是最常見的快取後端,redis-py 5.x 是最常見的 Python client。我們寫了一個可注入的 cache_aside 函式、用 Depends 注入 Redis、用 fakeredis 取代真 Redis 寫測試。沒有 Redis 的環境可以用 lru_cache(記憶體快取)或 fakeredis(純 Python 模擬器)替代。最後我們提醒了四個踩雷:序列化失敗、cache stampede、忘記 TTL、不一致,並把效能最佳化的方向放在延伸閱讀。今天結束時你的專案裡應該有一個「可開關」的快取層,CI 跑測試時用 fakeredis,正式部署時切到真 Redis。
結語
今天學了「把讀取加速」的方法,明天 Web Day 21「排程任務:APScheduler 與 cron」會把時間軸拉進來:有些事不用每次有人按才做、而是要在固定的時間自動發生。APScheduler 3.11.x 能在 FastAPI 行程裡掛一個排程器;cron 是作業系統層級的排程。今天的 Redis 容器也可以繼續留著,明天會示範「每天凌晨清掉過期快取」的實務寫法。
把快取策略寫進測試:性能斷言的技巧
昨天我們用 pytest 寫功能測試,今天可以再多寫一個「性能斷言」:用 pytest-benchmark 或簡單的時間量測,驗證「第二次讀取比第一次快」。
# tests/test_cache_perf.py
# 用簡單的時間量測驗證快取效果
import time
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_second_call_is_faster_with_cache(fake_redis):
# 第一次:應該從資料庫抓
start = time.perf_counter()
client.get("/items/1")
first_duration = time.perf_counter() - start
# 第二次:應該從快取抓
start = time.perf_counter()
client.get("/items/1")
second_duration = time.perf_counter() - start
# 快取應該比第一次快(雖然 fakeredis 也很慢)
assert second_duration <= first_duration * 1.5
這個測試不是嚴格的性能測試(要嚴格做要靠 pytest-benchmark),但能讓你快速驗證「快取真的有發揮作用」。如果你的服務用真的 Redis 做效能測試,這個斷言的差距會非常明顯:第一次讀資料庫可能要 20 毫秒,第二次讀快取只要 1 毫秒。
從單一快取到分層快取:地下一層的策略
真實的大型服務會有「多層快取」:瀏覽器、CDN、API gateway、應用程式快取、資料庫快取。今天寫的是應用程式這一層(Redis cache-aside),但它不該是唯一一層。常見的分層策略是:
- 瀏覽器:對靜態資源、不常變的 API 回應設
Cache-Control: max-age=...。 - CDN:把 GET 回應暫存在邊緣節點,降低應用程式負載。
- 應用程式快取(今天的主題):用 Redis 處理重複的計算或查詢。
- 資料庫查詢快取:PostgreSQL 的 shared_buffers、Redis 連線池、ORM 層的 identity map。
分層的關鍵是「越接近使用者、失效越快、容量越大」。瀏覽器層的失效時間可以拉到一天,CDN 層可能一小時,Redis 快取 30 秒到幾分鐘,資料庫層則視 query 結果而定。當你把這幾層都建好,從「使用者按按鈕到看到畫面」的延遲可以壓到 100 毫秒以內。今天只做 Redis 那一層,但觀念上要知道「這只是其中一環」。
選擇 TTL 的經驗法則
TTL 怎麼設是新手最常問的問題。實務上有幾條經驗法則:第一,TTL 與「資料變更頻率」成正比,使用者頭像、訂單狀態這類低頻更新的東西可以 TTL 30 分鐘;股價、即時庫存這類高頻更新的東西 TTL 5 秒就夠。第二,TTL 加上「隨機抖動」避免 stampede:實作上常寫成 ttl_seconds + random.randint(0, 30),讓過期時間分散在不同秒數。第三,「冷資料」的 TTL 可以短一點,因為就算失效也沒人會在意;「熱資料」的 TTL 可以長一點,但要記得搭配 invalidate。
另外一個常被忽略的設定是「負面快取」。當使用者查詢「id=9999」而這個 id 不存在,你當然不希望每次都去資料庫查「沒有」。把「查詢結果是 None」也寫進快取、TTL 短一點(例如 10 秒),就能避免惡意攻擊或 typo 流量打爆資料庫。Redis 把這個技巧叫做 cache negative results。
從單機到分散式:當你的服務長大
今天我們把所有快取邏輯放在「應用程式記憶體」或「單一 Redis」。當你部署多個 FastAPI 行程、Redis 仍然是單一節點時,這個架構運作得非常好:所有 FastAPI 行程共用同一份快取,命中率自然高。但當 Redis 變成 cluster(多節點),cache-aside 的失效順序要重新設計:「先更新資料庫、再刪快取」這個原則在分散式環境仍然適用,但要小心「刪完之後、另一個節點又把舊值寫回快取」的競態。
這個競態的標準解法是「延遲雙刪」:寫入資料庫後先刪一次快取、隔 500 毫秒再刪一次。這能確保「在這段時間內被其他節點寫進快取的舊值」會被清掉。當你的服務開始橫向擴展,這個技巧會變成必備。Web Day 31 討論 PostgreSQL 連線池時會再次觸及分散式議題,今天先記得「cache-aside 不只單機」。
今晚的練習:把昨天的端點加上快取
如果你從 Day 16 開始跟著做,現在的專案應該有同步端點、非同步端點、背景任務。今天最好的練習是「從昨天的測試裡挑一個端點,加上 cache_aside 包裝」。
具體步驟:在 app/cache.py 確認 cache_aside 函式可用;在 tests/conftest.py 確認 fake_redis fixture 存在;在某個 GET 端點裡把「讀資料庫」那段包成 loader、傳給 cache_aside;寫一個新的測試,連續呼叫兩次,確認第二次命中快取。整個過程大約 15 分鐘,跑完之後你會對 cache-aside 的實際運作有非常具體的感受。明天 APScheduler 也會延續同一個專案結構。
延伸資源
- Redis 官方文件(8,2025):
https://redis.io/docs/latest/ - redis-py 官方文件(5.2,2025):
https://redis.readthedocs.io/en/stable/ - fakeredis 官方說明(2.26,2025):
https://pypi.org/project/fakeredis/ - Caching Strategies 概論(AWS,2025):
https://aws.amazon.com/caching/best-practices/ - Cache-Aside Pattern(Microsoft Azure Architecture Center,2025):
https://learn.microsoft.com/en-us/azure/architecture/patterns/cache-aside
留言
張貼留言