Redis 深度实战：缓存、队列与分布式锁

Redis 远不止是一个 key-value 缓存。它是一个内存数据结构服务器，支持字符串、哈希、列表、集合、有序集合、流等多种数据结构，适用场景极其广泛。

1. 缓存策略

Cache-Aside Pattern（最常用）

import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_user(user_id: int):
    # 1. 先查缓存
    cached = r.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)
    # 2. 缓存未命中，查数据库
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    # 3. 写入缓存
    if user:
        r.setex(f"user:{user_id}", 3600, json.dumps(user))
    return user

缓存三大问题及解决方案

问题	描述	解决方案
缓存穿透	查询不存在的数据	布隆过滤器 / 缓存空值
缓存击穿	热点数据过期	互斥锁 (SETNX)
缓存雪崩	大量缓存同时过期	过期时间加随机值

# 缓存穿透：缓存空值
if user:
    r.setex(key, 3600, json.dumps(user))
else:
    r.setex(key, 60, "NULL")  # 缓存空值，防止穿透

# 缓存击穿：互斥锁
lock_key = f"lock:{key}"
if r.set(lock_key, "1", nx=True, ex=10):
    try:
        data = db.query_expensive()
        r.setex(key, 3600, data)
    finally:
        r.delete(lock_key)

# 缓存雪崩：随机过期时间
import random
ttl = 3600 + random.randint(0, 600)
r.setex(key, ttl, value)

2. 分布式锁

使用 Redis 实现可靠的分布式锁（Redlock 算法）：

import uuid

class RedisLock:
    def __init__(self, client, key, ttl=30):
        self.client = client
        self.key = f"lock:{key}"
        self.ttl = ttl
        self.token = str(uuid.uuid4())
    
    def acquire(self, retry=3):
        for _ in range(retry):
            if self.client.set(self.key, self.token, nx=True, ex=self.ttl):
                return True
            time.sleep(0.1)
        return False
    
    def release(self):
        # Lua 脚本保证原子性
        script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """
        self.client.eval(script, 1, self.key, self.token)

lock = RedisLock(r, 'inventory:product:123')
if lock.acquire():
    try:
        update_inventory(123)
    finally:
        lock.release()

3. 消息队列

List 实现简单队列

# 生产者
r.lpush('task_queue', json.dumps({'type': 'email', 'to': 'user@example.com'}))

# 消费者
while True:
    _, task = r.brpop('task_queue', timeout=5)
    if task:
        process_task(json.loads(task))

Stream 实现可靠队列

# Redis Stream 支持消费者组、ACK 确认、消息持久化
r.xadd('orders_stream', {'order_id': '12345', 'user_id': '678'}, maxlen=10000)

# 消费者组
r.xgroup_create('orders_stream', 'order_group', id='0', mkstream=True)
messages = r.xreadgroup('order_group', 'consumer1', {'orders_stream': '>'}, count=10)
for msg_id, data in messages[0][1]:
    process(data)
    r.xack('orders_stream', 'order_group', msg_id)

4. 排行榜

# Sorted Set 天生适合排行榜
r.zadd('leaderboard', {'user:1': 100, 'user:2': 200, 'user:3': 150})

# Top 10
top10 = r.zrevrange('leaderboard', 0, 9, withscores=True)

# 用户排名
rank = r.zrevrank('leaderboard', 'user:1')

# 分数范围查询
mid_range = r.zrangebyscore('leaderboard', 100, 200)

5. 性能优化建议

使用连接池：避免频繁创建和销毁连接
Pipeline 批量操作：减少网络往返次数
避免 KEYS 命令：使用 SCAN 替代 KEYS
设置合理的过期时间：防止内存无限增长
监控慢查询：SLOWLOG GET 10

总结

Redis 的核心价值在于内存级速度和丰富的数据结构。将 Redis 用作缓存、分布式锁、消息队列和排行榜，能够覆盖大多数后端开发中的常见需求。