当前位置：首页 > ai >正文

利用互斥锁或者利用逻辑过期解决缓存击穿问题

ai 2025/7/28 7:28:06

缓存击穿问题概述

缓存击穿是指某个 热点数据缓存过期 时，大量并发请求直接穿透缓存，同时访问数据库，导致数据库压力骤增甚至崩溃。以下是基于 互斥锁 和 逻辑过期 的解决方案：

一、缓存击穿的核心原因

热点数据失效：高并发场景下，某个热点数据的缓存突然过期，导致瞬时大量请求绕过缓存直接访问数据库。
无缓存保护：缓存未设计有效机制应对突发性并发穿透。

二、解决方案 1：互斥锁（Mutex Lock）

核心思想

当缓存失效时，仅允许一个线程查询数据库并重建缓存，其他线程阻塞等待，避免并发穿透。

实现步骤（以 Redis 分布式锁为例）

查询缓存
检查缓存是否存在，若存在则直接返回数据。

def get_data(key):data = redis.get(key)if data is not None:return data# 缓存未命中，尝试加锁return get_data_via_mutex(key)

获取分布式锁
使用 Redis 的 SETNX（或 Redlock 算法）实现分布式锁，确保原子性。

def acquire_lock(lock_key, expire=10):# 生成唯一标识（如UUID），避免误删其他线程的锁identifier = str(uuid.uuid4())if redis.set(lock_key, identifier, nx=True, ex=expire):return identifierreturn None

查询数据库并重建缓存
只有获取锁的线程执行数据库查询，其他线程等待。

def get_data_via_mutex(key):lock_key = f"lock:{key}"identifier = acquire_lock(lock_key)if not identifier:# 未获取锁，短暂等待后重试time.sleep(0.1)return get_data(key)try:# 再次检查缓存（可能已被其他线程更新）data = redis.get(key)if data:return data# 查询数据库data = db.query("SELECT * FROM table WHERE key=?", key)# 写入缓存redis.setex(key, 3600, data)finally:# 释放锁（需原子性操作）release_lock(lock_key, identifier)return data

释放锁
使用 Lua 脚本确保原子性释放锁：

-- KEYS[1]=lock_key, ARGV[1]=identifier
if redis.call("get", KEYS[1]) == ARGV[1] thenreturn redis.call("del", KEYS[1])
elsereturn 0
end

优缺点

优点：强一致性，彻底避免缓存击穿。
缺点：性能有损耗（线程需等待锁），锁超时时间需合理设置（过长影响并发，过短可能死锁）。

三、解决方案 2：逻辑过期（Logical Expiration）

核心思想

缓存数据不依赖物理过期时间，而是在数据中存储逻辑过期时间。当数据过期时，由单个线程异步更新缓存，其他线程继续返回旧数据。

实现步骤

缓存数据结构设计
在缓存中存储逻辑过期时间（expire_time）和实际数据：
```
{"data": "真实数据","expire_time": 1715000000  // 逻辑过期时间戳
}
```

查询缓存
检查逻辑过期时间，若未过期则直接返回数据。

def get_data(key):cache_data = redis.get(key)if cache_data:data_obj = json.loads(cache_data)if data_obj["expire_time"] > time.time():return data_obj["data"]else:# 触发异步更新async_refresh_cache(key)return data_obj["data"]  # 返回旧数据else:# 缓存完全不存在（需初始化）return load_data_and_init_cache(key)

异步更新缓存
使用互斥锁或标记位，确保仅一个线程执行数据库查询。

def async_refresh_cache(key):lock_key = f"refresh_lock:{key}"if redis.set(lock_key, 1, nx=True, ex=10):try:# 查询数据库new_data = db.query("SELECT * FROM table WHERE key=?", key)# 更新逻辑过期时间（例如延长1小时）new_expire_time = time.time() + 3600cache_obj = {"data": new_data, "expire_time": new_expire_time}redis.setex(key, 3600 * 24, json.dumps(cache_obj))  # 物理过期时间更长finally:redis.delete(lock_key)

初始化缓存
若缓存完全不存在（如服务重启），直接加载数据：

def load_data_and_init_cache(key):# 加锁防止并发初始化lock_key = f"init_lock:{key}"identifier = acquire_lock(lock_key)if not identifier:time.sleep(0.1)return get_data(key)try:# 再次检查缓存cache_data = redis.get(key)if cache_data:return json.loads(cache_data)["data"]# 查询数据库并写入缓存data = db.query("SELECT * FROM table WHERE key=?", key)expire_time = time.time() + 3600cache_obj = {"data": data, "expire_time": expire_time}redis.setex(key, 3600 * 24, json.dumps(cache_obj))return datafinally:release_lock(lock_key, identifier)