Redis 應用案例：快取、佇列與速率限制

Backend2026-10-13TryQuickToolBox

你已經將 Redis 加入你的技術堆疊，但只把它當作簡單的鍵值儲存使用。Redis 提供強大的資料結構與指令，能優雅地解決快取、佇列與速率限制問題。本指南將展示如何有效實作這些模式。

為何 Redis 適合這些應用案例？

Redis 是一種記憶體內資料儲存，以其速度與多功能性聞名。它支援字串、雜湊、列表、集合、有序集合等。其原子操作與內建過期機制，使其成為快取、佇列與速率限制的理想選擇。此外，它易於與任何程式語言整合。

使用 Redis 進行快取

快取是最常見的 Redis 應用案例。你將頻繁存取的資料存放於 Redis，以減少資料庫負載並加速回應。

基本的 Cache-Aside 模式

在 cache-aside 模式中，你的應用程式會先檢查 Redis。如果資料不存在（快取未命中），則從資料庫擷取，然後存入 Redis 以供下次使用。

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)

def get_user(user_id):
    cache_key = f'user:{user_id}'
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)
    user = fetch_user_from_db(user_id)  # 你的資料庫呼叫
    r.setex(cache_key, 3600, json.dumps(user))  # 快取 1 小時
    return user

最佳實踐：

一律設定過期時間（TTL），以避免資料過時。
使用有意義的鍵名，例如 user:123:profile。
處理快取失效：當資料變更時，更新或刪除快取項目。
考慮使用 Redis 雜湊來儲存物件的多個欄位，以節省記憶體。

進階快取模式

對於高流量應用程式，請考慮以下模式：

寫入穿透（Write-through）： 同時更新快取與資料庫。
寫入後送（Write-behind）： 先更新快取，再非同步更新資料庫。
預先重新整理（Refresh-ahead）： 在快取項目過期前主動重新整理。

使用 Redis 建立佇列

Redis 列表與串流非常適合建立簡單的訊息佇列。使用列表處理基本的工作佇列，使用串流處理更複雜的事件處理。

使用列表的簡單工作佇列

使用 LPUSH 新增工作，並使用 BRPOP 處理它們。BRPOP 會阻塞直到有工作可用，這對工作者來說很有效率。

# 生產者
r.lpush('jobs', json.dumps({'task': 'send_email', 'to': 'user@example.com'}))

# 消費者（工作者）
while True:
    _, job_data = r.brpop('jobs', timeout=30)
    job = json.loads(job_data)
    process_job(job)

最佳實踐：

為不同的工作類型或優先級使用不同的列表。
為失敗的工作實作死信佇列（dead-letter queue）。
使用 BRPOPLPUSH（或 Redis 6.2+ 中的 BLMOVE）以原子方式將工作移至處理中列表，確保可靠性。
使用 LLEN 監控佇列長度，以調整工作者數量。

使用 Redis Streams 進行事件溯源

Redis Streams（於 Redis 5.0 引入）提供更穩健的佇列，具備消費者群組、確認機制與重播功能。使用 XADD 新增事件，並使用 XREADGROUP 消費它們。

# 新增事件
r.xadd('events', {'type': 'order_placed', 'order_id': '123'})

# 消費者群組
r.xgroup_create('events', 'order_processors', id='0', mkstream=True)
while True:
    events = r.xreadgroup('order_processors', 'worker1', {'events': '>'}, count=10, block=5000)
    for event in events:
        process_event(event)
        r.xack('events', 'order_processors', event.id)

使用 Redis 進行速率限制

速率限制可保護你的 API 免於濫用，並確保公平使用。Redis 的原子操作使其成為計算請求數量的理想選擇。

固定視窗速率限制

為每個使用者遞增計數器並設定過期時間。如果計數超過限制，則封鎖請求。

def is_allowed(user_id, limit=100, window=60):
    key = f'rate_limit:{user_id}'
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    return current