$ loading_
帮助开发者掌握 Redis 生产级数据结构模式、缓存与并发控制方案
复制安装指令，让 AI 自动完成配置 · 推荐新手
请帮我安装 askskill 上的 "redis-patterns" 技能： 1. 下载 https://raw.githubusercontent.com/affaan-m/ECC/main/skills/redis-patterns/SKILL.md 2. 保存为 ~/.claude/skills/redis-patterns/SKILL.md 3. 装好后重载技能，告诉我可以用了
请为一个高并发电商商品详情服务设计 Redis 缓存方案，包含缓存键设计、TTL 策略、缓存穿透/击穿/雪崩防护，以及失效后的回源流程，并给出示例代码。
一套可落地的 Redis 缓存架构建议，附键命名规范、过期策略与示例代码。
请说明如何用 Redis 实现安全的分布式锁，包含加锁、续期、释放锁、防止误删，以及在多实例部署下的注意事项，并提供 Java 或 Node.js 示例。
分布式锁的实现思路、风险点说明，以及可直接参考的示例代码。
请基于 Redis 设计 API 限流和 Pub/Sub 消息通知方案，分别说明适用场景、核心数据结构、性能注意事项和连接管理最佳实践。
限流与消息通信的方案对比、实现建议，以及生产环境连接管理要点。
Quick reference for Redis best practices across common backend use cases.
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.
| Use Case | Structure | Example Key |
|---|---|---|
| Simple cache | String | product:123 |
| User session | Hash |
session:abc |
| Leaderboard | Sorted Set | scores:weekly |
| Unique visitors | Set | visitors:2024-01-01 |
| Activity feed | List | feed:user:456 |
| Event stream | Stream | events:orders |
| Counters / rate limits | String (INCR) | ratelimit:user:123 |
| Bloom filter / 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 hour
return product
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))
# Tag-based invalidation — group related keys under a 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)
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
-- 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
…
为 Quarkus 项目执行发布前验证闭环，涵盖构建、测试、扫描与差异审查。
帮助开发者掌握 Laravel 生产级架构模式与核心后端实现方法。