Redis 缓存实战：从入门到高并发优化

Redis 核心数据类型

String（字符串）

SET user:1:name "张三"
GET user:1:name
SETNX lock:resource 1      # 分布式锁
INCR counter               # 原子自增
SETEX session:abc 3600 "data"  # 带过期时间

Hash（哈希）

HSET user:1 name "张三" age 25 email "zhang@example.com"
HGET user:1 name
HGETALL user:1
HINCRBY user:1 age 1

List（列表）

LPUSH queue:task "task1" "task2"
RPUSH queue:task "task3"
LPOP queue:task
LRANGE queue:task 0 -1

Set（集合）

SADD tags:article1 "python" "redis" "cache"
SMEMBERS tags:article1
SINTER tags:article1 tags:article2  # 交集

ZSet（有序集合）

ZADD leaderboard 100 "player1" 200 "player2"
ZREVRANGE leaderboard 0 9  # 前 10 名
ZINCRBY leaderboard 10 "player1"

缓存策略

Cache-Aside（旁路缓存）

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

Write-Through（写穿透）

def update_user(user_id, data):
    # 同时更新数据库和缓存
    db.update("users", user_id, data)
    redis.setex(f"user:{user_id}", 3600, json.dumps(data))

Write-Behind（写回）

# 先更新缓存，异步写入数据库
redis.set(f"user:{user_id}", json.dumps(data))
async_queue.push(lambda: db.update("users", user_id, data))

缓存穿透解决方案

布隆过滤器

from redis import Redis
from redblom import RedBlom

redis = Redis()
blom = RedBlom(redis, "url:bloom")

# 添加 URL
blom.add("http://example.com/1")

# 检查是否存在
if blom.contains("http://example.com/1"):
    # 可能存在，查缓存/数据库
    pass
else:
    # 一定不存在，直接返回
    return None

缓存空值

def get_data(key):
    data = redis.get(key)
    if data == "__NULL__":
        return None
    if data:
        return json.loads(data)
    
    # 数据库查询
    data = db.query(key)
    
    # 即使为空也缓存，防止穿透
    if not data:
        redis.setex(key, 300, "__NULL__")
    else:
        redis.setex(key, 3600, json.dumps(data))
    
    return data

缓存雪崩解决方案

# 设置随机过期时间
import random

def set_cache(key, data, base_ttl=3600):
    # 添加随机偏移（±10%）
    jitter = random.randint(-360, 360)
    ttl = base_ttl + jitter
    redis.setex(key, ttl, json.dumps(data))

缓存击穿解决方案

互斥锁

import time

def get_user_with_lock(user_id):
    key = f"user:{user_id}"
    lock_key = f"lock:{key}"
    
    data = redis.get(key)
    if data:
        return json.loads(data)
    
    # 尝试获取锁
    if redis.setnx(lock_key, 1):
        redis.expire(lock_key, 10)  # 10 秒过期
        try:
            # 双重检查
            data = redis.get(key)
            if data:
                return json.loads(data)
            
            # 查数据库
            data = db.query(user_id)
            redis.setex(key, 3600, json.dumps(data))
            return data
        finally:
            redis.delete(lock_key)
    else:
        # 等待锁释放
        time.sleep(0.1)
        return get_user_with_lock(user_id)

高并发优化

管道批量操作

pipe = redis.pipeline()
for i in range(1000):
    pipe.set(f"key:{i}", f"value:{i}")
pipe.execute()  # 一次网络往返

连接池

from redis import ConnectionPool, Redis

pool = ConnectionPool(
    host="localhost",
    port=6379,
    max_connections=50,
    decode_responses=True
)
redis = Redis(connection_pool=pool)

监控与调优

# 查看内存使用
INFO memory

# 查看慢查询
SLOWLOG GET 10

# 实时监控
MONITOR

# 键空间统计
INFO keyspace

Redis 是高性能系统的基石，用好它能解决 80% 的性能问题！