1. 为什么我要自己封装一套缓存保护工具先交代一下背景。我负责的一个老项目用户量不算特别大但QPS的波动非常夸张运营一搞活动流量直接翻十几倍。某个核心商品详情页的接口下游数据库扛不住当时第一版解决办法很简单粗暴——加Redis缓存所有热点数据都往里塞过期时间统一设成30分钟。上线第一周一切正常第二周开始出问题。运营在某个深夜手动清了缓存第二天早高峰直接被打穿。监控面板上一片飘红数据库连接数瞬间打满慢查询堆积接口耗时从50ms直接飙到1500ms紧接着就是一连串的报警短信。当时排查下来问题是典型的缓存穿透和缓存击穿混在一起爆发。穿透是因为大量请求在查一个根本不存在的商品ID缓存里没有数据库里也没有所有请求都直接怼到数据库击穿是因为热key失效的瞬间几千个并发请求同时发现缓存没有于是一起去数据库捞数据。那段时间我每天都在修这个问题修完穿透又出击穿修完击穿又发现代码里到处都是散落的缓存逻辑每个开发写的写法都不一样有的加锁有的没加锁有的用空值缓存有的直接不处理。后来我实在受不了就花了两天时间把这两类问题统一封装成了一个工具库所有业务方接入同一个组件参数统一、策略统一、降级逻辑也统一。这篇文章就聊聊这个封装工具我是怎么设计的核心解决什么问题以及在真实业务里踩过的那些坑。先说清楚两个概念因为很多人把穿透和击穿混为一谈。缓存穿透指的是查询一个根本不存在的数据缓存和数据库都没有请求直接穿透缓存打到数据库。恶意攻击、异常ID、用户乱传参数都会导致这种情况特征是缓存永远不可能命中数据库压力是持续性的。缓存击穿指的是某个热点key在过期的瞬间大量并发请求同时发现缓存失效一起去数据库重建缓存数据库瞬时压力暴增。特征是在某一个时间点上集中爆发过了这个点就恢复了。还有一个容易混淆的是缓存雪崩那是大量key同时失效或者Redis节点整体宕机导致的系统性灾难和击穿的影响范围不在一个量级。这三者对应策略也完全不同。穿透必须靠空值缓存、布隆过滤器、参数校验来做前置拦截击穿需要靠互斥锁、逻辑过期、热点永不过期来控制并发重建的规模。一个大一统的封装必须同时覆盖这两种完全不同的解决路径并且要能够根据业务场景灵活切换。这正是我设计这个工具时面临的第一个挑战——不是代码写不出来而是接口设计必须足够清晰否则业务方不愿意用或者用错了配置。2. 工具的整体设计思路先定义清楚边界封装这件事最常见的失败原因不是功能不够而是边界模糊。业务方拿到一个工具不知道什么时候该用哪一个方法用错了反而更糟。所以我在动手写代码之前先做了两件事一是梳理出所有需要暴露的能力清单二是把工具类使用场景尽量控制在缓存读取数据和手动失效key两个动作上收敛API数量。最终我确认了工具需要具备的核心能力自动处理缓存未命中时的数据加载调用方只需要提供一个查询数据库的匿名函数内置互斥锁解决击穿场景下多个线程同时重建缓存的问题支持空值缓存解决穿透场景下大量无效请求打到数据库的问题支持自定义过期时间并且区分物理过期和逻辑过期两种模式提供手动删除缓存key的接口兼容运营清缓存、数据变更主动失效等场景默认加载兜底数据防止缓存重建过程中出现极端情况时接口直接报错注意在设计过程中我一直提醒自己这不是一个通用型缓存框架不需要像Spring Cache那样什么都管。它的核心目标就是解决缓存没命中的那一瞬间系统如何应对越聚焦越容易做好。API设计上最终只暴露了两个核心方法。第一个是查询接口的数据加载方法内部自动完成读缓存-缓存未命中-加锁重建-写入缓存-返回结果这一整套流程。第二个是主动失效方法业务方在更新数据库后调用确保下一次请求能拉到最新数据。class CacheProtector: 缓存穿透与击穿保护工具 def get_with_cache(self, key, loader, expire300, empty_cacheTrue, use_lockTrue, fallbackNone, need_random_delayTrue) - object: 核心查询方法 Args: key: 缓存key loader: 数据库加载函数, 无参数或一个key参数 expire: 过期时间, 秒 empty_cache: 是否缓存空结果, 默认True use_lock: 是否使用互斥锁保护, 默认True fallback: 兜底值, 缓存重建失败或极端情况下返回 need_random_delay: 是否在重建缓存后加入随机延迟, 防止雪崩 pass def invalidate(self, key) - bool: 主动失效指定key pass有人可能会问为什么加载函数不接收参数因为我对key的拼接规则做了统一约定业务方在外面把参数拼进key里loader内部闭包捕获了这些参数不需要我这个组件再去识别参数列表省去了一大堆反射和动态调用的复杂度。这个决定让代码量少了一半而且不容易出错。同步锁的实现我一开始用的是Python标准库的threading.Lock。但很快发现这在实际分布式环境下有个致命缺陷它是进程内锁如果服务部署了多个实例每个实例各有一个锁那同一个key在机器A上重建缓存的同时机器B也在重建锁形同虚设。然后我切换到了Redis分布式锁。做法很简单——在重建缓存之前先尝试设置一个lock_key设置成功才允许继续走数据库查询设置失败就在极短时间内自旋等待然后重新检查缓存是否已经被其他节点填充了。Redis实现分布式锁的代码如下import redis import time import uuid class RedisLock: 基于Redis的简易分布式锁 def __init__(self, client: redis.Redis, lock_key: str, expire10): self.client client self.lock_key lock_key self.expire expire self.token str(uuid.uuid4()) def acquire(self, retry5, sleep0.1) - bool: 尝试获取锁支持重试 for _ in range(retry): # SET NX EX 保证原子性 if self.client.set(self.lock_key, self.token, nxTrue, exself.expire): return True time.sleep(sleep) return False def release(self): 释放锁需要校验token防止误删 # 通过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.lock_key, self.token)这个锁的释放逻辑是关键。我看过很多团队的实现直接调用client.delete(lock_key)这是有隐患的。如果锁因为超时被自动释放了但业务逻辑还在执行此时另一个请求拿到了新锁等第一个请求执行完去删除key的时候会把第二个请求的锁误删掉。我上面这段代码用token验证每个锁都是一个UUID只有当前锁的持有者才能删除保证了安全性。3. 核心逻辑拆解一个方法如何同时挡住穿透和击穿接下来进入正题看看这个get_with_cache方法内部是怎么工作的。我按照数据处理链路依次拆开讲解每一步都有它存在的理由。3.1 第一步先查缓存命中就返回流程的第一步走Redis GET如果拿到值直接返回。正常业务下大部分请求都在这一步结束耗时一般不超1ms。这里有个细节从Redis取回来的值需要序列化和反序列化。在Python里我用json但这个工具有个需求是缓存对象可能会嵌套有些字段是自定义的枚举类型直接json.dumps会报错。后来我统一要求业务方在loader函数里就返回可被序列化的基础类型不要把一个ORM模型直接丢进来。这个问题在老代码里很常见值得提醒一下。如果Redis返回了特殊标记的空值占位符那就说明缓存里存的是一个空结果说明之前有人查过这个key且数据库里没有数据这个key直接短路处理返回空列表或者None不用去走数据库也不需要加锁因为空值缓存已经解决了穿透的问题。下面的代码展示了空值占位符的判断逻辑EMPTY_FLAG __CACHE_EMPTY__ def get_value(key): cached redis_client.get(key) if cached is None: return (miss, None) if cached EMPTY_FLAG: return (empty, None) return (hit, json.loads(cached))3.2 第二步未命中时分布式锁控制并发重建既然缓存没有就得走数据加载路线。如果use_lock为True先尝试获取分布式锁。获取不到锁的请求不能立刻去打数据库否则击穿问题根本没解决。想想发生击穿的那个瞬间一个热key过期500个并发请求同时进来如果没有锁500次查询全部穿过Redis打到数据库。加了锁之后最多只有1个请求去查数据库其余499个等待。拿不到锁的线程怎么处理这里有两种常见策略。第一种是自旋等待。线程A获取锁失败后每隔50ms重新尝试读取缓存直到把缓存读出来或等待超时。因为我设计的原则是重建缓存的时间一定很快——只在缓存中放计算结果操作尽量精简那么等待是有价值的。自旋等待的逻辑如下def _wait_for_cache(key, timeout3.0): deadline time.time() timeout while time.time() deadline: value redis_client.get(key) if value is not None: return value time.sleep(0.05) return None第二种是直接降级返回。如果业务允许短暂返回旧数据那等待是非常划算的。但如果业务要求实时性极高、旧数据不允许返回那自旋等待命中新缓存可能是唯一选择。这个工具中我在默认配置下选择自旋等待2秒如果2秒后还没等到就返回兜底值。这是我个人推荐的组合既保证了大多数场景的实时性又用兜底值兜住了极端异常。3.3 第三步加载数据、填充缓存、随机延迟拿到锁的线程开始执行loader函数查数据库。数据库返回结果后分两种情况处理。如果结果非空正常写入缓存设置过期时间。如果结果为空把这个结果序列化成EMPTY标志写入缓存过期时间设为一个较短但不可忽视的区间。这里有个关键的细节我加了一个need_random_delay参数它的作用是数据写入缓存的同时给过期时间追加一个随机偏移量。比如设定30分钟实际是25到35分钟之间的一个随机值。这样做是为了避免整个缓存体系里出现大量key同时过期的情况。很多团队遇到的问题就是数据量大之后同一批次写入的key总在同一个时间点集体失效然后引发一波小范围雪崩。加了这个随机延迟后这个隐患被直接消除掉了。生成随机过期时间的逻辑如下from random import randint def get_expire_with_jitter(base300): 在基础过期时间上增加10%到30%的随机浮动 jitter randint(int(base * 0.1), int(base * 0.3)) return base jitter还有一点容易被忽略写入缓存一定要用SET命令带上过期时间不要分开执行SET和EXPIRE。我见过不少老代码是set完之后单独调expire中间一旦异常或进程崩溃key就成了永不过期那种情况比穿透更可怕。3.4 第四步兜底逻辑并非可有可无为什么要配fallback数据库也可能挂或者极端高并发下我们主动拒绝了一部分请求。如果这一层不做兜底用户看到的直接就是500错误页。我在封装中默认支持两层兜底第一层是工具内置的静态默认值比如商品服务可以返回一个空列表虽然页面没内容但接口是通的第二层是本地进程内存缓存把最近一次成功查到的结果在进程内存里存一份周期半小时左右Redis挂了还能用这部分旧数据扛一阵子。内存兜底的实现非常简单不必动用额外的中间件import threading class LocalFallback: 进程内兜底存储 def __init__(self, max_items500): self._data {} self._lock threading.Lock() self._max_items max_items def set(self, key, value, ttl1800): with self._lock: if len(self._data) self._max_items: # 简单淘汰清掉一半 self._data.clear() self._data[key] { value: value, expire_at: time.time() ttl } def get(self, key): with self._lock: item self._data.get(key) if not item: return None if time.time() item[expire_at]: del self._data[key] return None return item[value]这套本地兜底设计的关键在于它不是给所有key用的只对热点key做保留所以我限制了max_items数量避免内存被无意义地占满。写入时机在成功拿到数据库数据之后每拿到一次就顺带更新一次本地备份。4. 穿透拦截的进阶方案布隆过滤器要不要上刚才说用空值缓存能解决穿透但实际场景中空值缓存有一个明显问题如果有人恶意构造大量不同的无效ID每个ID都能生成一个空值缓存Redis里会堆满占位符内存暴涨。这种情况下布隆过滤器才是更优解。布隆过滤器本质上是一个非常节省空间的集合数据结构它把元素映射到M位的位数组上用K个哈希函数做映射。它会告诉你一个元素一定不在集合里还是可能不在集合里。用在缓存穿透场景里就是预热所有有效ID判断请求的ID是否属于有效集合如果判断为不存在直接拒绝查询。但我在封装时没有在一开始就引入布隆过滤器原因有两个。第一需要全量ID的预加载流程这个流程在小项目里不是天然有的要写脚本去扫描数据库把所有有效ID灌进去。第二布隆过滤器存在误判率而且误判方向是把一个无效值判断成有效这等于漏掉了一部分应该拦截的请求需要用兜底指标量化这个概率。对很多中小项目来说空值缓存已经能解决80%的问题布隆过滤器属于锦上添花。我提供的建议是分阶段演进。上线初期工具配置空值缓存就可以。运行一段时间发现空值缓存量增长太快、无效请求比例太高再引出布隆过滤器。布隆过滤器的实现我是用Python的pybloof-lite直接接入Redis把位数组存在Redis的bitmap上这样所有实例共享同一份过滤数据。# 布隆过滤器模块, 基于Redis bitmap class BloomFilter: def __init__(self, client, key, capacity100000, error_rate0.01): self.client client self.key key self.size self._calculate_size(capacity, error_rate) self.hashes self._calculate_hashes(capacity, error_rate) def add(self, item): for seed in self.hashes: index self._hash_index(item, seed) self.client.setbit(self.key, index, 1) def might_contain(self, item) - bool: for seed in self.hashes: index self._hash_index(item, seed) if not self.client.getbit(self.key, index): return False return True def _hash_index(self, item, seed): # 简单混合哈希, 实际生产建议用murmurhash return (hash(item) seed * 31) % self.size布隆过滤器唯一需要关注的是扩容问题。数据量增长后误判率会上升通常达到容量的80%就该重新生成过滤器了。重新生成的方式是在后台跑一个新key全量重新加载跑完后切流老key保留几天再删。这些都是工程细节但做不好会让这个方案落不了地。5. 热key的识别与保护逻辑过期是更彻底的解法上面聊的互斥锁方案解决的是失效瞬间的并发重建。但它有一个必须承认的短板——热key在过期的那一瞬间总归会有一段时间缓存是空的请求最终还是要穿透到数据库虽然只有一个请求穿透但数据库还是遭遇了一次重击。如果这个key是每秒上万次访问的超高热key就算只有一次打到数据库对数据库造成的影响也可能是显著的。解决这个短板业界通用的办法是逻辑过期。逻辑过期的核心思路是物理上不让key过期只在value中存储一个逻辑过期时间戳。每次读取时判断逻辑时间戳是否过期——如果没过期直接返回如果过期了先返回旧值同时触发一个后台线程去重建缓存重建完成后新值覆盖旧值。这样打满整个生命周期没有哪个瞬间缓存内容为空数据库的压力被分摊到一个后台异步任务上而不是暴露在高频请求链路上。我把逻辑过期也封装在了同一个工具里给get_with_cache加了一个cache_strategy参数默认PHYSICAL可选LOGICAL。class CacheStrategy: PHYSICAL physical LOGICAL logical class HotKeyWrapper: 逻辑过期热key存储结构 staticmethod def pack(data, logical_expire): return json.dumps({ data: data, expire_at: time.time() logical_expire }) staticmethod def unpack(payload): obj json.loads(payload) if time.time() obj[expire_at]: return obj[data], True # 数据有效但已逻辑过期 return obj[data], False读取逻辑变得非常简单payload redis_client.get(key) if payload is None: # 布隆或空值判断后的真实未命中 return load_and_rebuild(key) data, expired HotKeyWrapper.unpack(payload) if not expired: return data # 逻辑过期了, 先返回旧值, 后台触发重建 threading.Thread(targetrebuild_cache, args(key, loader)).start() return data这套方案有个天然的大坑如果所有请求读到逻辑过期后都同时发起一个后台线程瞬间会有成百上千个重建任务数据库照样被打爆。所以逻辑过期方案必须配合分布式锁一起用。我让后台线程在启动前先尝试获取锁拿到锁的重建拿不到锁的静默跳过因为总会有一个线程完成任务。这个设计在项目里验证下来非常稳定。用逻辑过期之后我需要额外关注的是内存占用。因为key不物理过期一些不常用的key也会常驻Redis所以这种策略只推荐给真正的高热key使用。怎么判断哪些是热key最简单的是用Redis的hotkeys参数查或者在工具里埋点记录每个key的访问次数超过阈值自动切到逻辑过期模式。6. 完整代码实战从工具类到项目中的接入示例理论部分差不多了这里给一份可以实际落地的完整代码。考虑到不同团队语言栈不同我用Python给一份参考实现但这个工具的思路换成Java、Go都完全适用核心是锁的获取、空值占位、延迟重建这三个角色。import json import threading import time import uuid from random import randint from typing import Callable, Optional import redis EMPTY_FLAG __CACHE_EMPTY__ class CacheProtector: def __init__(self, redis_client: redis.Redis): self.client redis_client self.local_fallback LocalFallback() self.lock_timeout 10 self.lock_retry 3 self.lock_sleep 0.05 def get_with_cache( self, key: str, loader: Callable[[], object], expire: int 300, empty_ttl: int 60, use_lock: bool True, use_logical_expire: bool False, logical_background_loader: bool True, fallback: object None, need_random_delay: bool True, ) - object: # 1. 读缓存 cached self._get_from_redis(key) if cached is not None: if cached EMPTY_FLAG: return fallback if fallback is not None else None if not use_logical_expire: return cached # 逻辑过期模式 data, expired HotKeyWrapper.unpack(cached) if not expired: return data # 过期先返回旧值 if logical_background_loader and self._acquire_lock(key): threading.Thread( targetself._rebuild_cache, args(key, loader, expire), daemonTrue, ).start() return data # 2. 缓存未命中 if use_lock: # 等待其他线程/节点重建 waited self._wait_for_cache(key) if waited is not None: return waited if not self._acquire_lock(key): # 拿不到锁, 降级返回 return self._get_local_fallback(key, fallback) # 3. 执行真正的数据加载 try: result loader() if result is None or (hasattr(result, __len__) and len(result) 0): self._set_to_redis(key, EMPTY_FLAG, empty_ttl) return fallback if fallback is not None else None # 缓存重建完成后写入 ttl get_expire_with_jitter(expire) if need_random_delay else expire self._set_to_redis(key, result, ttl) self._set_local_fallback(key, result) return result except Exception: # 数据库异常时使用兜底 local_value self._get_local_fallback(key, fallback) if local_value is not None: return local_value raise def invalidate(self, key): self.client.delete(key) def _rebuild_cache(self, key, loader, expire): try: result loader() if result is not None and hasattr(result, __len__) and len(result) 0: self._set_to_redis(key, result, expire) finally: self._release_lock(key) def _acquire_lock(self, key): lock_key fcache_lock:{key} token uuid.uuid4().hex # 保存token到实例, 简化实现 self._current_token token for _ in range(self.lock_retry): if self.client.set(lock_key, token, nxTrue, exself.lock_timeout): return True time.sleep(self.lock_sleep) return False接入业务代码时只需要替换掉原来的先查缓存、查不到再查库里的散装逻辑改成一行调用哪怕团队里其他人水平参差不齐只要按这个方法写不会出大问题。举个例子。一个商品详情接口原来这么写def get_product_detail(product_id): key fproduct:{product_id} cached redis_client.get(key) if cached: return json.loads(cached) product db.query_one(select * from product where id %s, product_id) if product: redis_client.set(key, json.dumps(product), ex300) return product改成工具调用后protector CacheProtector(redis_client) def get_product_detail(product_id): key fproduct:{product_id} return protector.get_with_cache( keykey, loaderlambda: db.query_one(select * from product where id %s, product_id), expire300, fallback{product_id: product_id, name: 商品加载中, status: 0}, )改动成本极小但防穿透和防击穿能力直接补齐了。走读代码的人看到一行调用也能立刻明白这个数据是被缓存保护着的不会出现两个开发写两套缓存逻辑的混乱局面。7. 实测对比同一个接口接入前后的压力表现工具封装完我先在自己的测试环境里压了一轮然后拉到预发布环境做了业务验证。压测工具用的wrk模拟的场景是800并发持续压一个商品详情接口数据预热完毕后手动删除热key观察缓存失效瞬间的表现。接入前的结果缓存失效瞬间数据库QPS冲到2100数据库连接池打满接口P99耗时3420ms大约4秒后Redis缓存重建完成数据库压力回落期间出现少量数据库查询超时错误接入后的结果同样手动删除热key数据库QPS峰值稳定在50上下因为只有拿到分布式锁的那一个请求在查数据库其余并发请求在等待缓存重建接口P99耗时从3420ms回落到180ms左右无数据库错误无超时整轮压测下来Redis的读写命中率保持正常数据库的连接数和慢查询数对比非常直观。接入前慢查询数每分钟接近200条接入后降到了0到2条之间。这个结果其实在我的预期内但真正让我意外的是空值缓存在业务侧的收益——打开日志看了一轮竟然后台有3.2%的请求在查询根本不存在的商品ID这些请求在接入前全部穿到了数据库接入后直接短路在Redis层基本零成本拦截。8. 参数配置和运营经验不是每个key都适合同一套参数封装工具最大的陷阱就是一把钥匙开所有锁。我在内部推广这个工具的时候要求每个业务方必须根据自己接口的特征填参数不能全部用默认值。这里列一些我在实际配置中沉淀的经验值供参考。业务场景过期时间是否缓存空值是否用锁兜底策略商品详情页10-30分钟是是静态默认商品用户订单列表5分钟是是空列表库存数量30秒否是本地最近值配置类数据1小时否否静态配置参数不是设置一次就一劳永逸的。我建议每隔一段时间看访问热点分布把访问量top 100的key单独拎出来手动配置更短的逻辑过期策略把数据库的压力摊平到更细的时间颗粒度。还有缓存过期时间不要设太长。很多开发图省事把过期时间设成24小时想着反正有缓存就行但业务一旦数据更新频繁用户看到的旧数据会导致投诉。我的习惯是最长不超过1小时宁可多进几次Redis重建也不要让数据库在深夜遭受无意义的压力。9. 真实踩坑记录封装过程中遇到的三个大坑最后分享三个我在开发这个工具时真正花时间解决的问题这些是文档里基本不会写的东西。第一个坑是锁的粒度问题。一开始我的锁key是按用户维度设计的比如lock:user:{id}结果出现了一个异常场景某个用户短时间内发起了大量请求查询不同的商品锁全都落在on同一个用户key上导致不同的商品查询互相阻塞接口平均耗时直线上升。后来我改成锁和缓存key保持一致也就是lock:{product:123}只锁商品123这样锁的粒度和数据粒度一一对应互不干扰。第二个坑是本地兜底存储的并发问题。当地内存Fallback用了一个简单的字典在高并发写入的时候出现了部分数据丢失的现象。后来我仔细查了代码发现问题出在我对字典的读写不是原子的。添加threading.Lock后解决这算是个低级失误但确实容易踩特别是没经历过高并发场景的开发者。第三个坑更隐蔽关于Redis连接池的配置。因为工具内部会频繁地发起先GET、再SETNX、再GET等多轮Redis操作如果连接池max_connections配置过小在高并发下会出现大量连接等待报错。当时一个接口接入后服务端疯狂报redis connection pool exhausted排查了半天才意识到不是工具逻辑问题而是连接池参数需要调大。这件事提醒我封装工具不只是写对逻辑更要提醒使用者评估自身的基础设施配置。10. 封装工具的部署和扩展方向因为这个工具的本质是一个纯库不依赖特定框架所以我把它打成了一个独立的Python包内部通过setup.py分发团队里各个服务按需引入。这个包不包含任何业务逻辑只包含缓存保护、锁、空值占位、兜底四块核心能力未来如果要接入Go或者Java服务同样的设计思路可以直接平移。后续我还打算在这个工具上增加两个能力。一是自动监控和告警当某个key在单位时间内的未命中次数超过阈值时自动通知对应的开发负责人二是多级缓存的支持在Redis前面再挂一层本地进程内缓存把最热的那部分请求完全挡在Redis之外。多级缓存在压测中通常能把QPS再降一个量级但带来的数据一致性问题也需要仔细权衡等实现有结果了再写一篇展开聊。如果你现在的项目也在被缓存穿透或击穿问题困扰试试这套思路先把你散落的缓存代码统一收口到一个工具里再把锁、空值占位、逻辑过期这些能力逐项加上去比你在每个业务接口里单独补丁要省力得多而且后面维护的人会感谢你。
