《Python编程实战》8.1 Redis 缓存层次与键设计

import json
import fakeredis

redis = fakeredis.FakeRedis(decode_responses=True)
_local: dict[str, dict] = {}       # 进程内一级缓存

def get_profile(uid: int) -> dict:
    key = f"app:v1:user:{uid}"
    if key in _local:              # 1. 本地命中，零网络开销
        return _local[key]
    cached = redis.get(key)        # 2. 分布式缓存
    if cached is not None:
        _local[key] = json.loads(cached)
        return _local[key]
    row = {"id": uid, "name": f"user{uid}"}   # 3. 回源
    redis.set(key, json.dumps(row), ex=300)
    _local[key] = row
    return row

def invalidate(uid: int) -> None:
    key = f"app:v1:user:{uid}"
    _local.pop(key, None)          # 两级都要清
    redis.delete(key)

print(get_profile(7), "本地命中:", "app:v1:user:7" in _local)
invalidate(7)
print("失效后 本地:", "app:v1:user:7" in _local, " Redis:", redis.exists("app:v1:user:7"))

{'id': 7, 'name': 'user7'} 本地命中: True
失效后 本地: False  Redis: 0

import json
import fakeredis

redis = fakeredis.FakeRedis(decode_responses=True)
DB = {"user:42": {"id": 42, "name": "Ada", "plan": "pro"}}
db_calls = 0

def db_get_user(uid: int) -> dict:
    global db_calls
    db_calls += 1
    return DB.get(f"user:{uid}", {})

def get_user(uid: int) -> dict:
    key = f"app:v1:user:{uid}"
    cached = redis.get(key)                  # 1. 查缓存
    if cached is not None:
        return json.loads(cached)            # 命中
    row = db_get_user(uid)                   # 2. 未命中，回源
    redis.set(key, json.dumps(row), ex=300)  # 3. 回填，TTL 300s
    return row

print("第一次:", get_user(42), "db_calls =", db_calls)
print("第二次:", get_user(42), "db_calls =", db_calls)
print("TTL(s):", redis.ttl("app:v1:user:42"))

第一次: {'id': 42, 'name': 'Ada', 'plan': 'pro'} db_calls = 1
第二次: {'id': 42, 'name': 'Ada', 'plan': 'pro'} db_calls = 1
TTL(s): 300

redis.set("app:v1:cfg:theme", "dark")
redis.set("app:v1:cfg:lang", "zh")

pipe = redis.pipeline()
for k in ("app:v1:cfg:theme", "app:v1:cfg:lang"):
    pipe.get(k)
print("pipeline 批量取值:", pipe.execute())

pipeline 批量取值: ['dark', 'zh']

deleted = 0
for key in redis.scan_iter(match="app:v1:user:*"):
    deleted += redis.delete(key)
print("按前缀失效删除:", deleted, "键")

按前缀失效删除: 1 键

NULL = "__NULL__"
db_hits = 0

def get_with_null_cache(uid):
    global db_hits
    key = f"app:v1:user:{uid}"
    v = redis.get(key)
    if v is not None:
        return None if v == NULL else json.loads(v)
    db_hits += 1                  # 回源（数据库里也没有）
    redis.set(key, NULL, ex=60)   # 空值占位，挡住后续穿透
    return None

for _ in range(5):
    get_with_null_cache(999)
print("穿透防护: 5 次查询 -> DB 命中次数 =", db_hits)

穿透防护: 5 次查询 -> DB 命中次数 = 1

import time
import fakeredis

redis = fakeredis.FakeRedis(decode_responses=True)
db_calls = 0

def slow_db_load():
    global db_calls
    db_calls += 1
    time.sleep(0.05)          # 模拟慢查询
    return "hot-value"

def get_hot(key, ttl=300):
    if (v := redis.get(key)) is not None:
        return v
    if redis.set(key + ":lock", "1", nx=True, ex=5):   # 抢到锁的才回源
        try:
            value = slow_db_load()
            redis.set(key, value, ex=ttl)
            return value
        finally:
            redis.delete(key + ":lock")
    time.sleep(0.05)          # 没抢到锁：等锁持有者填好缓存
    return redis.get(key) or slow_db_load()

for _ in range(3):
    print("读:", get_hot("app:v1:hot"))
print("3 次并发读 -> 真正回源次数 =", db_calls)

读: hot-value
读: hot-value
读: hot-value
3 次并发读 -> 真正回源次数 = 1

import uuid
import fakeredis

redis = fakeredis.FakeRedis(decode_responses=True)

class Lock:
    def __init__(self, redis, name, ttl=10):
        self.redis, self.name, self.ttl = redis, name, ttl
        self.token = None

    def acquire(self) -> bool:
        token = uuid.uuid4().hex
        ok = self.redis.set(self.name, token, nx=True, ex=self.ttl)
        if ok:
            self.token = token
        return bool(ok)

    def release(self) -> bool:
        if self.token is None:
            return False
        with self.redis.pipeline() as pipe:
            while True:
                try:
                    pipe.watch(self.name)
                    if pipe.get(self.name) != self.token:
                        pipe.unwatch()
                        return False
                    pipe.multi()
                    pipe.delete(self.name)
                    pipe.execute()
                    return True
                except fakeredis.WatchError:
                    continue

lock = Lock(redis, "app:lock:order:1001", ttl=5)
print("A 获取:", lock.acquire())
print("B 获取(应失败):", Lock(redis, "app:lock:order:1001").acquire())
print("锁 TTL:", redis.ttl("app:lock:order:1001"))
print("A 释放:", lock.release())
print("释放后存在:", redis.exists("app:lock:order:1001"))
print("A 重复释放(应 False):", lock.release())

A 获取: True
B 获取(应失败): False
锁 TTL: 5
A 释放: True
释放后存在: 0
A 重复释放(应 False): False