Redis Patterns

affaan-m/ECC/skills/redis-patterns

作者 affaan-mef648e01899ba3e8dc6371642deaaf64b4477775無授權條款275K 個星標收錄於 2026年10月9日更新於 2026年10月9日儲存庫4 天前更新

Redis data structure patterns, caching strategies, distributed locks, rate limiting, pub/sub, and connection management for production applications. Use when adding caching, a distributed lock, rate limiting, or pub/sub with Redis, or when key design needs review.

僅含說明Software DevelopmentDevOps & Cloud
AI 產生的概覽

Redis 資料結構模式、快取、分散式鎖、限流、發布訂閱與連線管理的參考指南。

功能
提供後端常見情境下 Redis 最佳實務的速查參考，涵蓋資料結構選用、快取旁路與寫穿快取、快取失效、工作階段儲存、限流、分散式鎖、發布訂閱與 Streams、鍵命名與 TTL 策略、連線池、叢集與 Sentinel 設定、淘汰策略以及反模式。內容包含以 Python 與 Lua 撰寫的範例程式碼，以及將使用情境對應到資料結構、將模式對應到適用場合的表格。它產出的是指引與範例程式碼，本身不執行任何操作。
適用情境
適用於為應用程式加入快取、分散式鎖、限流、工作階段或權杖儲存，或使用 Redis 的發布訂閱與 Streams 時，也適用於檢視 Redis 鍵設計與正式環境設定。
執行需求
未隨附指令碼或工具，僅為說明與參考資料。範例假定已有 Redis 伺服器以及安裝 redis 用戶端函式庫的 Python 環境，部分模式另提到多節點鎖需使用 redlock-py 函式庫。

Redis Patterns

Quick reference for Redis best practices across common backend use cases.

How It Works

Redis is an in-memory data structure store that supports strings, hashes, lists, sets, sorted sets, streams, and more. Individual Redis commands are atomic on a single instance; multi-step workflows require Lua scripts, MULTI/EXEC transactions, or explicit synchronization to stay atomic. Data is optionally persisted via RDB snapshots or AOF logs. Clients communicate over TCP using the RESP protocol; connection pools are essential to avoid per-request handshake overhead.

When to Activate

  • Adding caching to an application
  • Implementing rate limiting or throttling
  • Building distributed locks or coordination
  • Setting up session or token storage
  • Using Pub/Sub or Redis Streams for messaging
  • Configuring Redis in production (pooling, eviction, clustering)

Data Structure Cheat Sheet

Use CaseStructureExample Key
Simple cacheStringproduct:123
User sessionHashsession:abc
LeaderboardSorted Setscores:weekly
Unique visitorsSetvisitors:2024-01-01
Activity feedListfeed:user:456
Event streamStreamevents:orders
Counters / rate limitsString (INCR)ratelimit:user:123
Bloom filter / HLLHyperLogLoghll:pageviews

Core Patterns

Cache-Aside (Lazy Loading)

python
import redisimport 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 hour    return product

Write-Through Cache

python
def update_product(product_id: int, data: dict):    # Write to DB first    db.execute("UPDATE products SET ... WHERE id = %s", product_id)
    # Immediately update cache    cache_key = f"product:{product_id}"    r.setex(cache_key, 3600, json.dumps(data))

Cache Invalidation

python
# Tag-based invalidation — group related keys under a setdef 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 Storage

python
import timeimport 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

Fixed Window (Simple)

python
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

Sliding Window (Lua — Atomic)

lua
-- sliding_window.lualocal 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    -- Use unique member (now + sequence) to avoid collisions within the same millisecond    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 1endreturn 0
python
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

Distributed Lock (Single Node — SET NX PX)

python
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)
# Usagetoken = acquire_lock("order:payment:123")if token:    try:        process_payment()    finally:        release_lock("order:payment:123", token)

For multi-node setups use the redlock-py library which implements the full Redlock algorithm.

Pub/Sub & Streams

Pub/Sub (Fire-and-Forget)

python
# Publisherdef publish_event(channel: str, payload: dict):    r.publish(channel, json.dumps(payload))
# Subscriber (blocking — run in separate thread/process)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 (Durable Queue)

python
# Producerdef emit(stream: str, event: dict):    r.xadd(stream, event, maxlen=10000)  # Cap stream length
# Consumer group — guarantees at-least-once deliverytry:    r.xgroup_create('events:orders', 'processor', id='0', mkstream=True)except Exception:    pass  # Group already exists
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)

Prefer Streams over Pub/Sub when you need delivery guarantees, consumer groups, or replay.

Key Design

Naming Conventions

# Pattern: resource:id:fielduser:123:profileorder:456:statuscache:product:789
# Pattern: namespace:resource:idmyapp:session:abc123myapp:ratelimit:user:123
# Pattern: resource:date (time-bound keys)stats:pageviews:2024-01-01

TTL Strategy

Data TypeSuggested TTL
User session24h (86400)
API response cache5–15 min
Rate limit windowMatch window size
Short-lived tokens5–10 min
Leaderboard1h–24h
Static/reference data1h–1 week

Always set a TTL. Keys without TTL accumulate indefinitely and cause memory pressure.

Connection Management

Connection Pooling

python
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

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

Sentinel (High Availability)

python
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

PolicyBehaviorBest For
noevictionError on write when fullQueues / critical data
allkeys-lruEvict least recently usedGeneral cache
volatile-lruLRU only among keys with TTLMixed data store
allkeys-lfuEvict least frequently usedSkewed access patterns
volatile-ttlEvict soonest-to-expirePrioritize long-lived data

Set via redis.conf: maxmemory-policy allkeys-lru

Anti-Patterns

Anti-PatternProblemFix
Keys with no TTLMemory grows unboundedAlways set TTL
KEYS * in productionBlocks the server (O(N))Use SCAN cursor
Storing large blobs (>100KB)Slow serialization, memory pressureStore reference + fetch from object store
Single Redis for everythingNo isolation between cache & queueUse separate DBs or instances
Ignoring connection pool limitsConnection exhaustion under loadSize pool to workload
Not handling cache miss stampedeThundering herd on cold startUse locks or probabilistic early expiry
FLUSHALL without thoughtWipes entire instanceScope deletes by key pattern

Cache Miss Stampede Prevention

python
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)  # Re-check after acquiring lock        if cached:            return json.loads(cached)        value = fetch_fn()        r.setex(key, ttl, json.dumps(value))        return value

Note: for multi-process deployments, replace the in-process lock with acquire_lock/release_lock from the Distributed Locks section above.

Examples

Add caching to a Django/Flask API endpoint: Use cache-aside with setex and a 5-minute TTL on the response. Key on the request parameters.

Rate-limit an API by user: Use fixed-window with pipeline(transaction=True) for low-traffic endpoints; use sliding-window Lua for accurate per-user throttling.

Coordinate a background job across workers: Use acquire_lock with a TTL that exceeds the expected job duration. Always release in a finally block.

Fan-out notifications to multiple subscribers: Use Pub/Sub for fire-and-forget. Switch to Streams if you need guaranteed delivery or replay for late consumers.

Quick Reference

PatternWhen to Use
Cache-asideRead-heavy, tolerate slight staleness
Write-throughStrong consistency required
Distributed lockPrevent concurrent access to a resource
Sliding window rate limitAccurate per-user throttling
Redis StreamsDurable event queue with consumer groups
Pub/SubBroadcast with no delivery guarantees needed
Sorted Set leaderboardRanked scoring, pagination
HyperLogLogApproximate unique count at low memory

Related

  • Skill: postgres-patterns — relational data patterns
  • Skill: backend-patterns — API and service layer patterns
  • Skill: database-migrations — schema versioning
  • Skill: django-patterns — Django cache framework integration
  • Agent: database-reviewer — full database review workflow

來源與署名

來源：affaan-m/ECC位於skills/redis-patterns提交ef648e0

授權條款： 無授權條款

內容歸原作者所有。SourceWeft 從公開儲存庫中收錄這些內容。

檢舉或申請下架

更多來自 affaan-m/ECC 的技能

Unified Notifications Ops

affaan-m

設計統一的通知策略，對來自 GitHub、Linear、鉤子與聊天工具的通知進行分類、路由、去重與升級。

Productivity & Workflow275K4 天前更新

Verification Loop

affaan-m

執行六階段程式碼驗證流程，並產出 PASS/FAIL 就緒報告。

Software Development275K4 天前更新

Ui To Vue

affaan-m

將批次 UI 設計截圖轉換為使用 Vant、Element Plus 或 Ant Design Vue 的 Vue 3 元件。

Software Development275K4 天前更新

Unified Memory

affaan-m

指導代理透過本機 ECC Memory Vault 在不同執行框架之間共享持久脈絡與交接資訊。

AI & Agents275K4 天前更新

Visa Doc Translate

affaan-m

透過 OCR 將簽證文件影像翻譯成英文，並產出原圖與譯文對照的雙語 PDF。

Documents & Office275K4 天前更新

Team Builder

affaan-m

互動式挑選器：探索可用的代理角色，讓使用者最多選五個，平行執行同一任務並彙整結果。

AI & Agents275K4 天前更新

更多Software Development技能

Playground

anthropics

精選

建立自成一體的互動式 HTML 遊樂場，包含控制項、即時預覽與可複製的提示詞輸出。

Software Development37K今天更新

M5 Onboard

anthropics

精選

透過 USB 偵測 M5Stack ESP32 開發板，燒錄 UIFlow 2.0 韌體並安裝 MicroPython 應用程式套件。

Software Development37K今天更新

Cardputer Buddy

anthropics

精選

指導迭代 Cardputer-Adv 的 MicroPython 應用程式套件：新增應用程式、透過 USB 序列埠推送檔案、查看日誌並執行 REPL 指令。

Software Development37K今天更新

Web Design Guidelines

vercel-labs

精選

依據從遠端來源取得的 Web Interface Guidelines 審查 UI 程式碼，並以 file:line 形式回報問題。

Software Development32K5 週前更新

Vercel React Native Skills

vercel-labs

精選

React Native 與 Expo 最佳實務規則，涵蓋清單效能、動畫、導覽、UI 模式、狀態與 monorepo 設定。

Software Development32K5 週前更新

Vercel React Best Practices

vercel-labs

精選

Vercel 的 React 與 Next.js 效能指南，以 70 條規則的形式用於撰寫、審查與重構程式碼。

Software Development32K5 週前更新