redis-patterns

Redis 模式

快速參考 Redis 在常見後端使用案例中的最佳實踐。

如何運作

Redis 是一個記憶體中的資料結構儲存庫，支援字串、雜湊、列表、集合、有序集合、串流等。單一實例上的個別 Redis 命令是原子性的；多步驟工作流程需要 Lua 腳本、MULTI/EXEC 交易或明確的同步機制來維持原子性。資料可選擇性地透過 RDB 快照或 AOF 日誌進行持久化。客戶端使用 RESP 協議透過 TCP 通訊；連線池對於避免每次請求的交握開銷至關重要。

何時啟用

• 為應用程式加入快取
• 實作速率限制或節流
• 建立分散式鎖定或協調
• 設定工作階段或令牌儲存
• 使用 Pub/Sub 或 Redis Streams 進行訊息傳遞
• 在生產環境中設定 Redis（池化、驅逐、叢集）

資料結構速查表

使用案例	結構	鍵值範例
簡單快取	字串 (String)	product:123
使用者工作階段	雜湊 (Hash)	session:abc
排行榜	有序集合 (Sorted Set)	scores:weekly
不重複訪客	集合 (Set)	visitors:2024-01-01
活動饋送	列表 (List)	feed:user:456
事件串流	串流 (Stream)	events:orders
計數器 / 速率限制	字串 (INCR)	ratelimit:user:123
布隆過濾器 / HLL	HyperLogLog	hll:pageviews

核心模式

快取旁路（惰性載入）

import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_product(product_id: int):
    cache_key = f"product:{product_id}"
    cached = r.get(cache_key)

    if cached:
        return json.loads(cached)

    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    r.setex(cache_key, 3600, json.dumps(product))  # TTL: 1 小時
    return product

寫入直通快取

def update_product(product_id: int, data: dict):
    # 先寫入資料庫
    db.execute("UPDATE products SET ... WHERE id = %s", product_id)

    # 立即更新快取
    cache_key = f"product:{product_id}"
    r.setex(cache_key, 3600, json.dumps(data))

快取失效

# 基於標籤的失效 — 將相關鍵分組在一個集合下
def cache_product(product_id: int, category_id: int, data: dict):
    key = f"product:{product_id}"
    tag = f"tag:category:{category_id}"
    pipe = r.pipeline(transaction=True)
    pipe.setex(key, 3600, json.dumps(data))
    pipe.sadd(tag, key)
    pipe.expire(tag, 3600)
    pipe.execute()

def invalidate_category(category_id: int):
    tag = f"tag:category:{category_id}"
    keys = r.smembers(tag)
    if keys:
        r.delete(*keys)
    r.delete(tag)

工作階段儲存

import time
import uuid

def create_session(user_id: int, ttl: int = 86400) -> str:
    session_id = str(uuid.uuid4())
    key = f"session:{session_id}"
    pipe = r.pipeline(transaction=True)
    pipe.hset(key, mapping={
        "user_id": user_id,
        "created_at": int(time.time()),
    })
    pipe.expire(key, ttl)
    pipe.execute()
    return session_id

def get_session(session_id: str) -> dict | None:
    data = r.hgetall(f"session:{session_id}")
    return data if data else None

def delete_session(session_id: str):
    r.delete(f"session:{session_id}")

速率限制

固定視窗（簡單）

def is_rate_limited(user_id: int, limit: int = 100, window: int = 60) -> bool:
    key = f"ratelimit:{user_id}:{int(time.time()) // window}"
    pipe = r.pipeline(transaction=True)
    pipe.incr(key)
    pipe.expire(key, window)
    count, _ = pipe.execute()
    return count > limit

滑動視窗（Lua — 原子性）

-- sliding_window.lua
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)

if count < limit then
    -- 使用唯一成員 (now + sequence) 以避免在同一毫秒內發生碰撞
    local seq_key = key .. ':seq'
    local seq = redis.call('INCR', seq_key)
    redis.call('EXPIRE', seq_key, math.ceil(window / 1000))
    redis.call('ZADD', key, now, now .. '-' .. seq)
    redis.call('EXPIRE', key, math.ceil(window / 1000))
    return 1
end
return 0

sliding_window = r.register_script(open('sliding_window.lua').read())

def allow_request(user_id: int) -> bool:
    key = f"ratelimit:sliding:{user_id}"
    now = int(time.time() * 1000)
    return bool(sliding_window(keys=[key], args=[now, 60000, 100]))

分散式鎖定

分散式鎖定（單一節點 — SET NX PX）

import uuid

def acquire_lock(resource: str, ttl_ms: int = 5000) -> str | None:
    lock_key = f"lock:{resource}"
    token = str(uuid.uuid4())
    acquired = r.set(lock_key, token, px=ttl_ms, nx=True)
    return token if acquired else None

def release_lock(resource: str, token: str) -> bool:
    release_script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    result = r.eval(release_script, 1, f"lock:{resource}", token)
    return bool(result)

# 使用方式
token = acquire_lock("order:payment:123")
if token:
    try:
        process_payment()
    finally:
        release_lock("order:payment:123", token)

對於多節點設定，請使用 redlock-py 函式庫，它實作了完整的 Redlock 演算法。

Pub/Sub 與 Streams

Pub/Sub（發送後忘記）

# 發布者
def publish_event(channel: str, payload: dict):
    r.publish(channel, json.dumps(payload))

# 訂閱者（阻塞 — 在單獨執行緒/程式中執行）
def subscribe_events(channel: str):
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    for message in pubsub.listen():
        if message['type'] == 'message':
            handle(json.loads(message['data']))

Redis Streams（持久佇列）

# 生產者
def emit(stream: str, event: dict):
    r.xadd(stream, event, maxlen=10000)  # 限制串流長度

# 消費者群組 — 保證至少一次傳遞
try:
    r.xgroup_create('events:orders', 'processor', id='0', mkstream=True)
except Exception:
    pass  # 群組已存在

def consume(stream: str, group: str, consumer: str):
    while True:
        messages = r.xreadgroup(group, consumer, {stream: '>'}, count=10, block=2000)
        for _, entries in (messages or []):
            for msg_id, data in entries:
                process(data)
                r.xack(stream, group, msg_id)

當需要傳遞保證、消費者群組或重播時，請偏好使用 Streams 而非 Pub/Sub。

鍵值設計

命名慣例

# 模式：resource:id:field
user:123:profile
order:456:status
cache:product:789

# 模式：namespace:resource:id
myapp:session:abc123
myapp:ratelimit:user:123

# 模式：resource:date（時限鍵）
stats:pageviews:2024-01-01

TTL 策略

資料類型	建議 TTL
使用者工作階段	24 小時 (86400）
API 回應快取	5–15 分鐘
速率限制視窗	符合視窗大小
短期令牌	5–10 分鐘
排行榜	1 小時–24 小時
靜態 / 參考資料	1 小時–1 週

始終設定 TTL。沒有 TTL 的鍵會無限期累積並導致記憶體壓力。

連線管理

連線池

from redis import ConnectionPool, Redis

pool = ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=20,
    decode_responses=True,
    socket_connect_timeout=2,
    socket_timeout=2,
)

r = Redis(connection_pool=pool)

叢集模式

from redis.cluster import RedisCluster

r = RedisCluster(
    startup_nodes=[{"host": "redis-1", "port": 6379}],
    decode_responses=True,
    skip_full_coverage_check=True,
)

哨兵模式（高可用性）

from redis.sentinel import Sentinel

sentinel = Sentinel(
    [('sentinel-1', 26379), ('sentinel-2', 26379)],
    socket_timeout=0.5,
)
master = sentinel.master_for('mymaster', decode_responses=True)
replica = sentinel.slave_for('mymaster', decode_responses=True)

驅逐策略

策略	行為	最佳適用場景
noeviction	空間滿時寫入會出錯	佇列 / 關鍵資料
allkeys-lru	驅逐最近最少使用	通用快取
volatile-lru	僅在設有 TTL 的鍵中進行 LRU	混合資料儲存
allkeys-lfu	驅逐最不常使用	偏斜存取模式
volatile-ttl	驅逐即將到期的鍵	優先保留長壽命資料

透過 redis.conf 設定：maxmemory-policy allkeys-lru

反模式

反模式	問題	修正方法
無 TTL 的鍵	記憶體無限增長	始終設定 TTL
在生產環境中使用 KEYS *	阻塞伺服器 (O(N))	使用 SCAN 遊標
儲存大型 blob (>100KB)	序列化慢、記憶體壓力	儲存參考 + 從物件儲存庫擷取
用單一 Redis 處理所有事情	快取與佇列之間無隔離	使用不同的 DB 或實例
忽略連線池限制	負載下連線耗盡	根據工作負載調整池大小
未處理快取未命中雪崩	冷啟動時的驚群效應	使用鎖定或機率性提前過期
不加思索地使用 FLUSHALL	清除整個實例	按鍵模式範圍刪除

快取未命中雪崩預防

import threading

_locks: dict[str, threading.Lock] = {}
_locks_mutex = threading.Lock()

def get_with_lock(key: str, fetch_fn, ttl: int = 300):
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    with _locks_mutex:
        if key not in _locks:
            _locks[key] = threading.Lock()
        lock = _locks[key]
    with lock:
        cached = r.get(key)  # 取得鎖定後重新檢查
        if cached:
            return json.loads(cached)
        value = fetch_fn()
        r.setex(key, ttl, json.dumps(value))
        return value

注意：對於多程序部署，請用上方「分散式鎖定」章節中的 acquire_lock/release_lock 取代進程內鎖定。

範例

為 Django/Flask API 端點加入快取： 使用快取旁路模式，搭配 setex 並將回應的 TTL 設為 5 分鐘。鍵值基於請求參數。

依使用者對 API 進行速率限制： 對低流量端點使用固定視窗搭配 pipeline(transaction=True)；對精確的每位使用者節流使用滑動視窗 Lua。

協調跨工作者的背景任務： 使用 acquire_lock，其 TTL 應超過預期的任務持續時間。務必在 finally 區塊中釋放鎖定。

將通知扇出到多個訂閱者： 使用 Pub/Sub 進行發送後忘記。如果需要保證傳遞或為後續消費者提供重播功能，請切換至 Streams。

快速參考

模式	何時使用
快取旁路	讀取量大，可容忍些微過時
寫入直通	需要強一致性
分散式鎖定	防止對資源的並發存取
滑動視窗速率限制	精確的每位使用者節流
Redis Streams	具有消費者群組的持久事件佇列
Pub/Sub	廣播，無需傳遞保證
有序集合排行榜	排名計分、分頁
HyperLogLog	低記憶體下的近似不重複計數

相關

• 技能：postgres-patterns — 關聯式資料模式
• 技能：backend-patterns — API 與服務層模式
• 技能：database-migrations — 資料庫版本控制
• 技能：django-patterns — Django 快取框架整合
• 代理：database-reviewer — 完整的資料庫審查工作流程