缓存设计：这次怎么落地的

缓存设计 Redis 本地缓存 分布式缓存
作者: 醉月思 发布日期: 2021-10-14 分类: 技术实践

热点接口的 QPS 从几百涨到几千，最先顶不住的是数据库。我在应用里加了 Guava 本地缓存，命中率不错，但多实例部署后各节点各存一份，穿透和击穿的问题立刻冒出来。后面才逐步引入 Bloom Filter、Redis 和更完整的缓存架构。

本地缓存

为什么先用本地缓存

本地缓存最简单，不需要额外部署。

// 使用 Guava Cache
Cache<String, String> cache = CacheBuilder.newBuilder()
    .maximumSize(1000)
    .expireAfterWrite(10, TimeUnit.MINUTES)
    .build();

// 使用缓存
String value = cache.getIfPresent("key");
if (value == null) {
    value = loadFromDatabase("key");
    cache.put("key", value);
}

优势：

  • 速度快，无网络开销
  • 实现简单
  • 成本低

劣势：

  • 每个实例都有独立缓存，数据不一致
  • 内存有限
  • 无法共享缓存

本地缓存适用场景

适合：

  • 配置数据
  • 静态数据
  • 热点数据（且只读）
  • 单实例部署

不适合：

  • 多实例部署
  • 需要数据一致性
  • 缓存数据量大

Redis 分布式缓存

为什么需要 Redis

多实例部署后，本地缓存数据不一致，必须用分布式缓存。

# 部署 Redis
docker run -d --name redis -p 6379:6379 redis:latest

基础操作

import redis

# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 设置缓存
r.set('user:123', json.dumps({'name': 'Alice', 'age': 30}))
r.expire('user:123', 3600)  # 1 小时过期

# 获取缓存
data = r.get('user:123')
if data:
    user = json.loads(data)
else:
    user = load_from_database(123)
    r.set('user:123', json.dumps(user))
    r.expire('user:123', 3600)

缓存穿透

大量请求查询不存在的数据，缓存没命中，直接打到数据库。

解决：布隆过滤器

from pybloom_live import BloomFilter

# 初始化布隆过滤器
bf = BloomFilter(capacity=1000000, error_rate=0.001)

# 预加所有存在的 key
for key in all_keys:
    bf.add(key)

# 查询时先检查布隆过滤器
if key not in bf:
    return None  # 一定不存在，不需要查数据库

缓存击穿

热点 key 过期，大量请求同时打到数据库。

解决：互斥锁

import time
from threading import Lock

lock = Lock()

def get_with_lock(key):
    # 尝试获取锁
    if lock.acquire(blocking=False):
        try:
            value = r.get(key)
            if value is None:
                value = load_from_database(key)
                r.set(key, value)
                r.expire(key, 3600)
            return value
        finally:
            lock.release()
    else:
        # 没拿到锁，等待后重试
        time.sleep(0.1)
        return get_with_lock(key)

缓存雪崩

大量 key 同时过期，数据库压力骤增。

解决：随机过期时间

import random

def set_cache_with_random_ttl(key, value, base_ttl=3600):
    # 在基础时间上加随机偏移
    ttl = base_ttl + random.randint(0, 300)  # ±5 分钟
    r.set(key, value)
    r.expire(key, ttl)

缓存更新策略

Cache Aside Pattern

应用负责维护缓存一致性。

def update_user(user_id, user_data):
    # 1. 更新数据库
    db.update('users', user_id, user_data)
    
    # 2. 删除缓存
    r.delete(f'user:{user_id}')

Write Through Pattern

应用写入时同步更新缓存。

def update_user(user_id, user_data):
    # 1. 更新数据库
    db.update('users', user_id, user_data)
    
    # 2. 更新缓存
    r.set(f'user:{user_id}', json.dumps(user_data))
    r.expire(f'user:{user_id}', 3600)

Write Behind Pattern

应用写入后异步更新缓存。

def update_user(user_id, user_data):
    # 1. 更新数据库
    db.update('users', user_id, user_data)
    
    # 2. 异步更新缓存
    threading.Thread(
        target=lambda: (
            r.set(f'user:{user_id}', json.dumps(user_data)),
            r.expire(f'user:{user_id}', 3600)
        )
    ).start()

多级缓存

架构设计

graph TB A[请求] --> B{L1:本地缓存} B -->|命中| C[返回] B -->|未命中| D{L2:Redis} D -->|命中| E[更新L1] --> C D -->|未命中| F[数据库] F --> G[更新L2] --> E style A fill:#FFD700 style B fill:#90EE90 style D fill:#90EE90 style F fill:#FFB6C1

实现

class MultiLevelCache:
    def __init__(self):
        self.local_cache = {}
        self.redis = redis.Redis()
    
    def get(self, key):
        # L1: 本地缓存
        if key in self.local_cache:
            return self.local_cache[key]
        
        # L2: Redis 缓存
        value = self.redis.get(key)
        if value:
            self.local_cache[key] = value
            return value
        
        # L3: 数据库
        value = self.load_from_database(key)
        if value:
            self.redis.set(key, value)
            self.redis.expire(key, 3600)
            self.local_cache[key] = value
        
        return value
    
    def delete(self, key):
        self.local_cache.pop(key, None)
        self.redis.delete(key)

缓存一致性

最终一致性

缓存和数据库不需要强一致，最终一致即可。

def update_data(key, value):
    # 1. 更新数据库
    db.update(key, value)
    
    # 2. 发送消息到消息队列
    mq.send('cache_invalidate', {'key': key})

# 消费者
def process_message(message):
    key = message['key']
    cache.delete(key)

强一致性

需要使用分布式锁，确保缓存和数据库原子更新。

import redis.lock

def update_with_consistency(key, value):
    lock = redis.lock.RedisLock(redis, f'lock:{key}')
    
    with lock:
        # 1. 更新数据库
        db.update(key, value)
        
        # 2. 更新缓存
        cache.set(key, value)

踩过的坑

坑一：缓存过期时间没设好

一开始没设过期时间，缓存数据越来越多，Redis 内存满了。

解决：合理设置过期时间，定期清理。

# 设置合理的过期时间
r.set('user:123', json.dumps(user))
r.expire('user:123', 3600)  # 热点数据 1 小时
r.expire('config:system', 86400)  # 配置数据 1 天

坑二：大 Value 问题

缓存了很大的对象，导致 Redis 性能下降。

解决：

  • 拆分大对象
  • 只缓存必要字段
  • 使用压缩
# 只缓存必要字段
def get_user_summary(user_id):
    user = db.get_user(user_id)
    return {
        'id': user['id'],
        'name': user['name'],
        'avatar': user['avatar']
    }

# 使用压缩
import gzip
def set_compressed(key, value):
    compressed = gzip.compress(json.dumps(value).encode())
    r.set(key, compressed)

def get_compressed(key):
    compressed = r.get(key)
    return json.loads(gzip.decompress(compressed).decode())

坑三：缓存冷启动

系统重启后，缓存是空的，大量请求打到数据库。

解决：

  • 预热缓存
  • 限流保护
  • 灰度启动
# 预热缓存
def warm_up_cache():
    for user_id in all_user_ids:
        user = db.get_user(user_id)
        cache.set(f'user:{user_id}', json.dumps(user))
        cache.expire(f'user:{user_id}', 3600)

# 限流保护
from ratelimit import limits

@limits(calls=100, period=60)  # 每分钟最多 100 次调用
def get_user(user_id):
    return cache.get_or_load(f'user:{user_id}', lambda: db.get_user(user_id))

写在最后

缓存这东西，不是万能药，用不好反而会带来问题。

解决了：

  • 数据库压力
  • 响应速度
  • 吞吐量

带来了：

  • 一致性问题
  • 复杂度增加
  • 运维成本

实施之前先评估：

  • 数据读写比例
  • 数据一致性要求
  • 缓存命中率
  • 运维能力

不是所有场景都需要缓存，有时候优化数据库查询就够用。

这次缓存优化花了两周，从本地缓存到多级缓存，再到缓存一致性处理。优化完成后，数据库 QPS 从 1000 降到 200，响应时间从 500ms 降到 50ms。

版权声明: 本文首发于 指尖魔法屋-缓存设计：这次怎么落地的（https://blog.thinkmoon.cn/post/63-cache-design-local-distributed-architecture/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章