API 缓存策略计算引擎(专业版)
输入
is_cache_expired(now,created_at,ttl);jitter_ttl(base,jitter_pct=0.1,rng=None);cache_control(ttl,public=True);cache_strategy(db_found,is_hot_key=False,null_ttl_ratio=0.1)
输出
过期bool;抖动后TTL(int);Cache-Control字符串;(策略,说明,TTL系数)
硬性规则
- 恰好到期判过期;抖动可注入 random.Random(seed) 复现。
- db_found=False→CACHE_NULL;热点→MUTEX_REBUILD;其余STANDARD。
- 抖动结果 max(...,0) 不为负。
示例
is_cache_expired(60,0,60) → True
jitter_ttl(60,0.1,random.Random(1)) → 固定可复现值,且∈[54,66]
cache_strategy(False) → ("CACHE_NULL","空值缓存防穿透",0.1)
Python 引擎(可独立运行)
# -*- coding: utf-8 -*-
import random
def is_cache_expired(now, created_at, ttl):
"""当前时刻 >= 写入时刻+TTL 即判过期(恰好到期也算过期)。"""
return now >= created_at + ttl
def jitter_ttl(base_ttl, jitter_pct=0.1, rng=None):
"""防缓存雪崩:在基础TTL上叠加 ±jitter_pct 随机抖动。
rng 可注入 random.Random(seed) 保证结果可复现;结果不为负。"""
r = rng if rng is not None else random.Random()
delta = base_ttl * jitter_pct
final = int(base_ttl + r.uniform(-delta, delta))
return max(final, 0)
def cache_control(ttl, public=True):
"""生成 Cache-Control 响应头。"""
return f"Cache-Control: {'public' if public else 'private'}, max-age={ttl}"
def cache_strategy(db_found, is_hot_key=False, null_ttl_ratio=0.1):
"""依据DB结果与是否热点key给出缓存处置,返回(策略,说明,TTL系数)。
DB无数据→空值缓存防穿透;热点key未命中→互斥重建防击穿;其余标准缓存。"""
if not db_found:
return ("CACHE_NULL", "空值缓存防穿透", null_ttl_ratio)
if is_hot_key:
return ("MUTEX_REBUILD", "热点key互斥重建防击穿", 1.0)
return ("STANDARD", "标准缓存", 1.0)