redis-patterns

Name: redis-patterns
Author: affaan-m

Redis Patterns

針對常見後端使用場景整理的 Redis 最佳實踐快速指南。

運作原理

Redis 是基於記憶體的資料結構儲存庫，支援字串（strings）、雜湊（hashes）、列表（lists）、集合（sets）、排序集合（sorted sets）、串流（streams）等多種型別。Redis 針對單一實例的個別指令皆具備原子性（atomic）；若要實現多步驟的跨指令工作流程，則需要透過 Lua 腳本、MULTI/EXEC 事務（transactions）或明確的同步機制來維持原子性。資料可透過 RDB 快照或 AOF 日誌進行選擇性持久化。用戶端透過 TCP 使用 RESP 協定進行通訊；建置連線池（connection pool）是避免每次請求都產生握手開銷的關鍵。

時機與適用場景

為應用程式新增快取機制
實作 API 流量限制（Rate limiting）或調節（Throttling）
建置分散式鎖定（Distributed locks）或跨節點協調機制
設置 Session 狀態或 Token 儲存
使用 Pub/Sub 或 Redis Streams 進行訊息傳遞
配置正式環境的 Redis（連線池、記憶體淘汰策略、叢集架構）

資料結構速查表

使用場景	資料結構	Key 命名範例
簡單快取	String	`product:123`
使用者 Session	Hash	`session:abc`
排行榜	Sorted Set	`scores:weekly`
不重複訪客統計	Set	`visitors:2024-01-01`
活動動態牆	List	`feed:user:456`
事件串流	Stream	`events:orders`
計數器 / 限流器	String (INCR)	`ratelimit:user:123`
伯隆過濾器 / HLL	HyperLogLog	`hll:pageviews`

核心模式

Cache-Aside 旁路快取（延遲載入）

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

Write-Through 直寫快取

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))

快取無效化（Cache Invalidation）

# 基於 Tag 的失效機制 — 將相關 key 歸類至同一個 set 下
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)

Session 狀態儲存

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}")

流量限制（Rate Limiting）

固定視窗（固定時間區間，實作簡單）

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
    -- 使用唯一的成員標記（時間戳記 + 序號）以避免在同一毫秒內發生碰撞
    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]))

分散式鎖定（Distributed Locks）

分散式鎖定（單一節點 — 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 函式庫。

發布/訂閱與 Streams

Pub/Sub（發布/訂閱 — 即發即忘）

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

# 訂閱者 (Subscriber，阻塞式 — 建議於獨立執行緒/程序中執行)
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（可持久化的訊息佇列）

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

# 消費者組 (Consumer group — 保證至少交付一次 at-least-once)
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)

提示：當你需要訊息交付保證（Delivery guarantees）、消費者組（Consumer groups）或歷史訊息重放（Replay）功能時，請優先選擇 Streams 而非 Pub/Sub。

Key 設計

命名規範

# 格式：資源:ID:欄位
user:123:profile
order:456:status
cache:product:789

# 格式：命名空間:資源:ID
myapp:session:abc123
myapp:ratelimit:user:123

# 格式：資源:日期（具時效性的 key）
stats:pageviews:2024-01-01

TTL（過期時間）策略

資料類型	建議 TTL
使用者 Session	24 小時 (`86400`)
API 回應快取	5–15 分鐘
流量限制視窗	與限流視窗長度一致
短效 Token	5–10 分鐘
排行榜	1 小時–24 小時
靜態 / 參考資料	1 小時–1 週

請務必為所有 key 設定 TTL。未設定 TTL 的 key 會無上限累積，最終導致記憶體用盡。

連線管理

連線池（Connection Pooling）

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)

叢集模式（Cluster Mode）

from redis.cluster import RedisCluster

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

哨兵模式（Sentinel 高可用性）

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)

記憶體淘汰策略（Eviction Policies）

策略	行為描述	最佳適用場景
`noeviction`	記憶體滿時寫入操作直接報錯	佇列 / 核心關鍵資料
`allkeys-lru`	淘汰最久未使用的 key（LRU）	通用快取系統
`volatile-lru`	僅在設定了 TTL 的 key 中套用 LRU 淘汰	快取與持久化混合的儲存庫
`allkeys-lfu`	淘汰使用頻率最低的 key（LFU）	熱點存取分佈極度傾斜的場景
`volatile-ttl`	優先淘汰最快過期的 key	優先保留長效資料

透過 redis.conf 配置：maxmemory-policy allkeys-lru

反模式（應避免的做法）

反模式	問題所在	修正方案
Key 未設定 TTL	記憶體無止盡膨脹	務必設定 TTL
在正式環境執行 `KEYS *`	阻塞單執行緒伺服器（O(N) 複雜度）	改用 `SCAN` 遊標迭代查詢
儲存大型 Blob（>100KB）	序列化慢且造成記憶體壓力	僅存參照位址，本體存至物件儲存（Object Store）
單一 Redis 混用所有用途	快取與佇列等作業缺乏隔離性	使用獨立的資料庫（DB index）或實例
忽視連線池上限設定	高負載時耗盡連線數	依據工作負載妥善配置連線池
未處理快取雪崩/擊穿	快取失效或冷啟動時造成資料庫癱瘓	使用鎖定機制或機率性提前過期策略
隨意執行 `FLUSHALL`	清空整台 Redis 實例的所有資料	按 key 模式進行範圍刪除

防止快取擊穿/雪崩（Cache Miss Stampede Prevention）

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

提示：若為多程序（Multi-process）部署環境，請將程式中的本機鎖替換為前文「分散式鎖定」章節提供的 acquire_lock/release_lock。

實戰範例

為 Django/Flask API 端點新增快取：
採用 Cache-aside 模式，搭配 setex 設定 5 分鐘的 TTL。Key 則依據請求參數組合而成。

依據使用者對 API 進行流量限制：
對於流量較小的端點，可使用搭配 pipeline(transaction=True) 的固定視窗模式；若需精準調控單一使用者的請求頻率，請採用滑動視窗 Lua 腳本。

跨 Worker 節點協調背景任務：
使用 acquire_lock 並設定大於預期任務執行時間的 TTL，且務必在 finally 區塊中進行鎖定釋放。

扇出（Fan-out）通知至多個訂閱者：
即發即忘的場景可直接使用 Pub/Sub；若需要確保訊息可靠交付或供晚加入的消費者重放歷史訊息，請改用 Streams。

快速對照參考

模式	適用時機
Cache-aside	讀多寫少、可容忍微小的資料時效差
Write-through	需要強一致性（Strong consistency）
分散式鎖定	防止多個服務同時存取競爭資源
滑動視窗限流	精準控制單一使用者的請求頻率
Redis Streams	需要具備消費者組的可靠事件佇列
Pub/Sub	廣播訊息且無需訊息交付保證
Sorted Set 排行榜	排名計分、動態分頁
HyperLogLog	低記憶體開銷的超大基數不重複估算

您可能也喜歡

kubernetes-patterns

240Kbackend-api

Kubernetes workload patterns, resource management, RBAC, probes, autoscaling, ConfigMap/Secret handling, and kubectl debugging for production-grade deployments. Use when writing or reviewing Kubernetes manifests, or debugging probes, RBAC, autoscaling, or resource limits.