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周前更新