redis-patterns

Redis 模式

常見後端使用情境下 Redis 最佳實務的快速參考。

運作原理

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

何時啟用

• 為應用程式加入快取
• 實作速率限制或節流
• 建立分散式鎖或協調機制
• 設定工作階段或令牌儲存
• 使用發布/訂閱或 Redis 串流進行訊息傳遞
• 在生產環境中設定 Redis（連線池、淘汰機制、叢集）

資料結構速查表

使用案例	結構	範例鍵值
簡單快取	字串	product:123
使用者工作階段	雜湊	session:abc
排行榜	有序集合	scores:weekly
不重複訪客	集合	visitors:2024-01-01
活動饋送	列表	feed:user:456
事件串流	串流	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 + 序號）以避免在同一毫秒內發生衝突
    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 演算法的 redlock-py 函式庫。

發布/訂閱與串流

發布/訂閱（發送後即忘）

# 發布者
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 串流（持久佇列）

# 生產者
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)

當需要傳遞保證、消費者群組或重播功能時，請優先選擇串流而非發布/訂閱。

鍵值設計

命名慣例

# 模式: 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 處理所有事務	快取與佇列之間沒有隔離	使用不同的資料庫或執行個體
忽略連線池限制	負載下連線耗盡	根據工作負載調整池大小
未處理快取未命中雪崩	冷啟動時出現驚群效應	使用鎖或機率性提前過期
未經思考使用 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 端點加入快取： 使用 cache-aside 搭配 setex，對回應設定 5 分鐘 TTL。根據請求參數建立鍵值。

按使用者限制 API 速率： 對於低流量端點使用固定視窗與 pipeline(transaction=True)；對於精確的按使用者節流，使用滑動視窗 Lua。

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

將通知分發給多個訂閱者： 對於發送後即忘的情境使用發布/訂閱。如果需要保證傳遞或為較晚加入的消費者提供重播功能，請改用串流。

快速參考

模式	何時使用
Cache-aside	讀取密集，可容忍輕微過時
Write-through	需要強一致性
分散式鎖	防止對資源的並行存取
滑動視窗速率限制	精確的按使用者節流
Redis 串流	具有消費者群組的持久事件佇列
發布/訂閱	無需傳遞保證的廣播
有序集合排行榜	排名評分、分頁
HyperLogLog	以低記憶體使用率取得近似唯一計數

相關

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