- 内存缓存(进程内)
import time
from collections import OrderedDict
class MemoryCache:
definit(self, ttl=300, max_size=1000):
self.cache = OrderedDict()
self.ttl = ttl
self.max_size = max_size
def get(self, key): if key not in self.cache: return None value, ts = self.cache[key] if time.time() - ts > self.ttl: del self.cache[key] return None self.cache.move_to_end(key) # LRU return value def set(self, key, value): if len(self.cache) >= self.max_size: self.cache.popitem(last=False) # evict oldest self.cache[key] = (value, time.time())优点:速度最快(in-process 内存)
缺点:不跨进程、进程重启丢数据
- Redis 缓存(单层)
import redis, json
r = redis.Redis()
def search_serp_redis(query, ttl=300):
key = f"serp:{hash(query)}"
cached = r.get(key)
if cached:
return json.loads(cached)
data = call_serp(query) r.setex(key, ttl, json.dumps(data)) return data优点:跨进程、持久化、生态成熟
缺点:网络往返(0.5-1ms)、内存成本
- Memcached 缓存
import pymemcache
c = pymemcache.Client((‘localhost’, 11211))
def search_serp_memcached(query, ttl=300):
key = f"serp:{query}"
cached = c.get(key)
if cached:
return json.loads(cached)
data = call_serp(query) c.set(key, json.dumps(data), expire=ttl) return data优点:纯内存缓存,延迟 < 1ms
缺点:无持久化(重启丢)、无数据结构(LRU 淘汰)
- 多层缓存(L1 + L2)
import time
L1 = {} # 进程内
L2 = redis.Redis() # 共享
def search(query):
# L1:进程内,1 秒 TTL
if query in L1 and time.time() - L1[query][“ts”] < 1:
return L1[query][“data”]
# L2:Redis,5 分钟 TTL cached = L2.get(f"serp:{query}") if cached: data = json.loads(cached) L1[query] = {"data": data, "ts": time.time()} return data # Miss:回源 data = call_serp(query) L1[query] = {"data": data, "ts": time.time()} L2.setex(f"serp:{query}", 300, json.dumps(data)) return data优点:极快(命中 L1 50%) + 跨进程(命中 L2 30%) + 高可用
缺点:实现复杂
- 实测对比(我项目 30 天)
指标 内存 Redis Memcached 多层
命中率 35% 38% 32% 60%
命中率(L1 only) - - - 50%
命中率(L2 only) - - - 30%
平均延迟(命中) 0.1ms 0.8ms 0.4ms 0.2ms
失败率 5%(重启) 0.1% 1% 0.05%
月成本(单实例) 0 $5 $3 $5
实现复杂度 ★ ★★ ★★ ★★★ - 选型
场景 选
单进程 / 简单场景 内存
跨进程 / 标准场景 Redis
高 QPS / 低延迟 Memcached
高可用 / 多层 多层 - 4 个 SERP API 缓存的坑
坑 1:cache key 冲突
错:只 hash query,不同 region 拿到同一份
key = f"serp:{hash(query)}"
对:含所有影响结果的参数
key = f"serp:{hash(f’{query}|{gl}|{hl}|{num}')}"
坑 2:TTL 太久导致 stale
错:TTL 1 小时
r.setex(key, 3600, data)
对:不同类型不同 TTL
新闻类 1 分钟,排名类 1 小时,普通 5 分钟
坑 3:雪崩(热门 key 过期)
错:1000 个请求同时打 cache miss
def search_unsafe(query):
if r.get(query): return …
return call_serp(query) # 同时 1000 个调用
对:加 lock,只 1 个回源
def search_safe(query):
if r.get(query): return …
if not r.set(f"lock:{query}“, “1”, ex=5, nx=True):
time.sleep(0.1)
return r.get(query) or search_safe(query)
try:
return call_serp(query)
finally:
r.delete(f"lock:{query}”)
坑 4:内存缓存无限增长
错:无限增长
cache[key] = data
对:LRU eviction
if len(cache) > max_size:
cache.popitem(last=False) # LRU 淘汰
cache[key] = data
8. 实战数据(1 个月运行)
我项目用多层缓存跑 1 个月:
L1 命中 50%(进程内,1 秒 TTL)
L2 命中 30%(Redis,5 分钟 TTL)
Miss 20%(回源 + 写缓存)
总调用 95,000 → 实际 SERP API 调用 19,000
节省 80% SERP API 调用
月成本 $5.7
与 serpbase 集成的 4 个优势
auto-refund 100% 触发:缓存层 miss 回源失败,serpbase 退 credit,缓存层不亏
失败不重试缓存:缓存 miss 时如果 SERP 失败,缓存 null 1 分钟(避免雪崩)
preheat 热点查询:启动时把热点 query 提前查一次,缓存到 L1 + L2
监控命中率:低于 30% 触发告警,可能 cache key 设计有问题