redis-lessons.md 74 KB
Redis 缓存
术核 AI · 后端工程课件
01 连接 Redis、选择类型并设置 TTL
在隔离的 Redis 7 练习实例中操作字符串、哈希和集合，建立可辨认的键名与过期规则。
本节目标
- 通过环境变量配置带密码的 Redis 连接
- 区分字符串、哈希与集合的操作
- 解释 TTL 的 -1、-2 和覆盖写入行为
必要概念
缓存副本与业务真源
本模块以MySQL 8.4 LTS的InnoDB notes_lab为笔记真源，Redis只保存可重新生成的副本。缓存丢失影响读取性能，不应删除真实笔记。业务连接显式RC、UTC与有界等待；数据库命令行课另行演示默认RR。计数器、锁和消息队列的正确性要求不同，不能自动套用可随时丢弃的缓存策略。
数据类型决定命令
字符串可保存序列化 JSON；Hash 保存多个字段；Set 表示无重复成员。不同类型有各自命令，GET 一个 Hash 会返回 WRONGTYPE。这里不使用额外 RedisJSON 模块，JSON 文本由 Python 编解码。
TTL 是数据生命期
TTL 正数表示剩余秒数，-1 表示存在但没有过期时间，-2 表示键不存在。SET 默认覆盖值并移除原 TTL，除非显式设置新期限或保留期限。缓存写入通常把值和 EX/PX 放在同一条 SET 中，避免写值成功却未设置过期。
动手实现
1. 启动只绑定本机端口的练习实例
需要已安装并运行 Docker、Python 3.11+。下面是 Linux / macOS Bash，在 notes-service 执行，只创建本机专用 notes-redis。config 文件含练习密码，目录只供当前用户访问；挂载单个配置文件供容器读取。生产应按凭据管理和 TLS 方案部署，不能直接使用这份本机配置。
文件：终端 / 当前课时
mkdir -p redis-lab
chmod 700 redis-lab
export REDIS_PASSWORD="$(python3 -c 'import secrets; print(secrets.token_urlsafe(32))')"
python3 - <<'PY'
import os
from pathlib import Path
password = os.environ['REDIS_PASSWORD']
p = Path('redis-lab/redis.conf')
p.write_text('bind 0.0.0.0\nprotected-mode yes\nport 6379\n'
f'requirepass {password}\n'
'save ""\nappendonly no\nmaxmemory 128mb\nmaxmemory-policy allkeys-lru\n')
p.chmod(0o644)
PY
printf 'redis-lab/\n' >> .gitignore
docker run --name notes-redis -d -p 127.0.0.1:6380:6379 \
-v "$PWD/redis-lab/redis.conf:/usr/local/etc/redis/redis.conf:ro" \
redis:7 redis-server /usr/local/etc/redis/redis.conf
export REDIS_HOST=127.0.0.1 REDIS_PORT=6380
python -m pip install 'redis>=5.3,<6'
python -m pip freeze > requirements-lock.txt
2. 建立后续复用的客户端
保存为 redis_client.py。REDIS_PASSWORD 必须存在，忘记设置时应尽早失败。客户端启动不会立即验证连通，main 中 PING 才实际访问。这里使用 redis-py 5.3 系列稳定 API，Redis 7 不支持的较新命令不在示例中使用。
文件：redis_client.py
import os
import redis
r = redis.Redis(
host=os.environ.get("REDIS_HOST", "127.0.0.1"),
port=int(os.environ.get("REDIS_PORT", "6380")),
password=os.environ["REDIS_PASSWORD"],
db=0, decode_responses=True,
socket_connect_timeout=0.2, socket_timeout=0.2,
health_check_interval=30, max_connections=16,
)
if __name__ == "__main__":
print("ping", r.ping())
print("version", r.info("server")["redis_version"])
r.close()
3. 操作三种数据类型
保存为 redis_types.py，执行 python redis_types.py。所有键使用 notes:lab: 前缀，仅删除本课明确列出的三个键。反复练习不用 FLUSHALL；不同业务键不能共用同一个名字。
文件：redis_types.py
from redis_client import r
keys = ["notes:lab:types:title", "notes:lab:types:hash", "notes:lab:types:tags"]
r.delete(*keys)
r.set(keys[0], "first note", ex=60)
r.hset(keys[1], mapping={"title": "first note", "pinned": "false"})
r.expire(keys[1], 60)
r.sadd(keys[2], "python", "api", "python")
r.expire(keys[2], 60)
print(r.get(keys[0]))
print(r.hgetall(keys[1]))
print(sorted(r.smembers(keys[2])))
print([r.type(k) for k in keys])
r.delete(*keys)
4. 验证 TTL 与覆盖写入
保存为 redis_ttl.py 并运行。等待只发生在专用 1 秒练习键，不用长时间阻塞服务。SET 后 TTL 为 -1 的一幕用于认识风险；最终删除该练习键。
文件：redis_ttl.py
import time
from redis_client import r
k = "notes:lab:ttl-probe"
r.set(k, "v1", ex=1)
print("expires", r.ttl(k))
time.sleep(1.1)
print("missing", r.ttl(k))
r.set(k, "v2", ex=60)
r.set(k, "v3")
print("no expiration", r.ttl(k))
r.set(k, "v4", ex=60)
print("restored expiration", r.ttl(k))
r.delete(k)
验收结果
- PING 返回 True，服务版本属于 Redis 7。
- 字符串、Hash、Set 的值和类型符合预期，Set 中 python 只出现一次。
- 能复现 TTL=-2 与 TTL=-1，所有长期缓存写入都明确设置期限。
生产约束与故障演练
笔记详情需要缓存，但重启 Redis 不能影响 MySQL 中的真实笔记。
实现决策
- 键名包含应用、环境与数据用途。
- 固定开发端口并设置认证。
- 缓存写入与 TTL 在同一条命令中完成。
- MySQL连接使用MYSQL_*配置，宿主3307/容器3306；Redis端口仍为6380。
隔离环境演练
在专用键上移除 TTL 并观察 -1，再修复为带 EX 的 SET。
恢复验收
- 客户端没有明文硬编码密码。
- 练习不使用 FLUSHALL。
- 能从键名和 TYPE 识别数据用途。
常见错误
NOAUTH 或 AuthenticationError
原因：客户端未提供 REDIS_PASSWORD，或重开终端后环境变量丢失。
处理：从本机受保护配置恢复正确密码到当前环境，确认客户端指向 6380；不把密码贴入日志。
WRONGTYPE Operation against a key holding the wrong kind of value
原因：相同键名先被写成一种类型，后被另一类命令操作。
处理：用 TYPE 检查，修改键名命名空间；仅清理确认属于当前实验的键。
扩展练习
写入一个 30 秒 JSON 文本缓存，并证明覆盖写入时必须重新处理 TTL。
提示：JSON 先 json.dumps，再 SET ex=30；不要分开 SET 与 EXPIRE。
参考答案：
r.set("notes:lab:json-probe", json.dumps({"schema":1,"id":1}), ex=30)。再次 r.set 同键时仍带 ex=30，或明确使用 keepttl=True；读后用 json.loads。SET 未指定过期且未 KEEPTTL 会移除已有 TTL。
技术文档
02 实现 Cache-Aside 与带版本的 JSON 缓存
先查 Redis，未命中再读 MySQL，验证第一次回源与第二次命中的完整路径。
本节目标
- 实现显式命中、未命中和回填
- 把数据库类型转成稳定 JSON 结构
- 用 schema 区分缓存格式版本与业务数据版本
必要概念
Cache-Aside 的读路径
应用先 GET；不存在时查询数据库，再把结果 SET 到缓存。Redis 不会自动知道 MySQL 更新。先把基本路径做对，后续逐课处理失效、并发回填和故障降级。
两种版本解决不同问题
缓存外层schema=1表示JSON协议，note.version是MySQL记录的业务版本。MySQL不会自动把DATETIME附带UTC信息，也不会把TINYINT(1)自动转成JSON布尔值，所以序列化边界显式转换。将来添加JSON列时还应确认驱动返回的是文本还是已解码对象，不能让多个入口各自猜测。
缓存值也需要验证
反序列化 JSON 后还要检查结构和编号，坏缓存应被当作未命中处理。数据库时间转换为带时区 ISO 8601，不用 pickle 反序列化来自缓存的任意对象。本课缓存只包含列表出的笔记字段。
动手实现
1. 准备一条MySQL笔记
先完成数据库第03课mysql_client.py；其中PyMySQL[rsa]>=1.1,<2、DictCursor、autocommit=False、显式UTC/RC与连接超时已配置。设置MYSQL_HOST=127.0.0.1、MYSQL_PORT=3307、MYSQL_DATABASE=notes_lab、MYSQL_USER=app_user和本机MYSQL_PASSWORD；不输出密码。notes从初始DDL就包含version。REDIS_PASSWORD保持Redis第01课配置。运行脚本后设置export NOTE_ID=输出的实际编号。
文件：seed_cache_note.py
from mysql_client import connect
conn=connect()
try:
with conn.cursor() as cur:
cur.execute("INSERT INTO notes(title,content) VALUES(%s,%s)",
("cache-lab","database is the source"))
note_id=cur.lastrowid
conn.commit()
print("NOTE_ID=",note_id)
except BaseException:
conn.rollback()
raise
finally:
conn.close()
2. 显式转换MySQL结果并实现Cache-Aside
保存note_cache.py，完整替换旧实现。PyMySQL读取DATETIME(6)通常得到无tzinfo的datetime，本项目确认所有写入都采用UTC后才附加timezone.utc；BOOLEAN的0/1显式转bool。正文上限16000字符，与API保持一致。读取成功也commit结束RC事务，异常rollback，finally关闭。此教学回源每次新建连接，生产改用数据库第10课的有界池或API模块对应异步连接池。
文件：note_cache.py
import json
from datetime import datetime,timezone
from mysql_client import connect
from redis_client import r
def key(note_id):
return f"notes:lab:v1:note:{{{note_id}}}"
def utc_iso(value):
if not isinstance(value,datetime):
raise TypeError("expected database datetime")
# DATETIME没有自带时区；此赋值依赖所有写入均为UTC的项目约定。
if value.tzinfo is None:
value=value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc).isoformat()
def normalize_note(row):
if row is None:
return None
return {
"id":int(row["id"]),"title":row["title"],"content":row["content"],
"pinned":bool(row["pinned"]),
"created_at":utc_iso(row["created_at"]),
"updated_at":utc_iso(row["updated_at"]),
"version":int(row["version"]),
}
def load_note(note_id):
conn=connect()
try:
with conn.cursor() as cur:
cur.execute("SELECT id,title,content,pinned,created_at,updated_at,version FROM notes WHERE id=%s",(note_id,))
row=cur.fetchone()
conn.commit()
return normalize_note(row)
except BaseException:
conn.rollback()
raise
finally:
conn.close()
def encode(note):
return json.dumps({"schema":1,"note":note},ensure_ascii=False)
def decode(raw,note_id):
data=json.loads(raw)
if not isinstance(data,dict) or data.get("schema")!=1:
raise ValueError("unsupported cache schema")
note=data["note"]
if not isinstance(note,dict) or type(note.get("id")) is not int or note["id"]!=note_id:
raise ValueError("invalid cached note id")
if not isinstance(note.get("title"),str) or not 1<=len(note["title"])<=120:
raise ValueError("invalid cached title")
if not isinstance(note.get("content"),str) or len(note["content"])>16000:
raise ValueError("invalid cached content")
if type(note.get("pinned")) is not bool or type(note.get("version")) is not int or note["version"]<1:
raise ValueError("invalid cached fields")
for field in ("created_at","updated_at"):
parsed=datetime.fromisoformat(note[field])
if parsed.tzinfo is None or parsed.utcoffset().total_seconds()!=0:
raise ValueError("cached timestamp must be UTC")
return note
def get_note(note_id):
raw=r.get(key(note_id))
if raw is not None:
try:
return decode(raw,note_id),"cache"
except (ValueError,KeyError,TypeError):
r.delete(key(note_id))
note=load_note(note_id)
if note is not None:
r.set(key(note_id),encode(note),ex=60)
return note,"database"
3. 对照两次读取来源
保存为 cache_read.py。只清除当前练习笔记的一个缓存键。第一次来源 database，第二次 cache，内容应相等，TTL 在 1 到 60 之间。没有记录时返回 None，本课尚未缓存不存在的结果。
文件：cache_read.py
import os
from redis_client import r
from note_cache import key, get_note
note_id = int(os.environ["NOTE_ID"])
r.delete(key(note_id))
first, source1 = get_note(note_id)
second, source2 = get_note(note_id)
assert first == second and first is not None
print(source1, source2, r.ttl(key(note_id)))
print(first)
4. 注入坏格式并验证自动回源
先写入不可解析文本，再写入不受支持的 schema。每次调用都应返回 database 和合法笔记，然后缓存被合法 JSON 替换。Redis 网络错误尚未在此函数内处理，第 06 和 09 课补齐有界降级。
文件：cache_bad_format.py
import os
from redis_client import r
from note_cache import key, get_note
note_id = int(os.environ["NOTE_ID"])
for bad in ["not-json", '{"schema":99,"note":{"id":1}}']:
r.set(key(note_id), bad, ex=60)
note, source = get_note(note_id)
print(source, note["id"])
assert source == "database"
验收结果
- 第一次返回 database，第二次返回 cache，业务字段一致。
- 缓存含schema和整数note.version；pinned是真正JSON布尔值，时间为+00:00的ISO8601。
- 坏 JSON 和未知 schema 不直接导致业务解码崩溃，回源后得到有效内容。
生产约束与故障演练
同一笔记被反复读取，希望降低数据库访问同时保留格式升级能力。
实现决策
- MySQL 成功读取后才回填。
- 缓存格式版本和业务版本分离。
- 异常与不存在使用不同结果。
隔离环境演练
给单个练习键写入坏 JSON 和未知 schema，检查回源和重新编码。
恢复验收
- 命中来源可观测。
- 无任意对象反序列化。
- 缓存值不包含未设计对外展示的敏感字段。
常见错误
缓存返回整数pinned或没有时区的时间
原因：直接序列化驱动结果，没有把MySQL类型转成API约定的DTO。
处理：在normalize_note中集中bool、整数与UTC时间转换；对DATETIME补UTC依赖明确存值约定，不能猜测已有未知来源时间。
修改代码增加字段后旧缓存导致 KeyError
原因：缓存格式没有版本和校验，部署后仍读取旧结构。
处理：提升缓存 schema 或键名前缀，允许旧键自然过期；读取方明确接受的结构。
扩展练习
让记录不存在时也能返回明确来源，并说明为什么不能把数据库异常当成不存在。
提示：None 表示成功查询且无行；连接异常代表没有完成查询。
参考答案：
load_note成功查询且fetchone为None时，返回None与database来源。PyMySQL连接异常、语句超时和数据转换异常继续传播，不能伪造不存在并写负缓存。读取事务也显式commit或rollback并关闭；故障降级不意味着吞掉真源错误。
技术文档
03 提交后失效缓存，并复现旧值回填竞态
把失效放到事务提交之后，按固定时序制造旧快照覆盖，明确最终一致性的边界。
本节目标
- 正确安排数据库提交与缓存删除的顺序
- 复现先读旧值后回填的竞态
- 设计失效失败后的重试与一致性验收
必要概念
先提交数据库，再失效缓存
数据库提交成功后删除详情缓存，下一次读取从真源重建。如果在提交前删除，另一个读请求可能读到尚未更新的数据库旧值并重新缓存。提交后删除减少这一类窗口，但不消除所有并发竞态。
旧读取可以晚于新写入回填
读者 A 在写者提交前读到 v1，写者 B 提交 v2 并删除缓存，A 随后把 v1 回填。即使删除发生在提交后，缓存仍可能短暂陈旧。TTL 只限制一次缓存副本的存活时间，不应被表述为完整一致性证明。
失效失败是待补偿工作
MySQL事务和Redis命令没有共同本地事务。数据库已提交、DEL失败不能报告回滚。生产可在同一InnoDB事务写业务行和Outbox（事件表提前由独立DDL创建），后台可靠重试删除；若要求立即读取自己提交的结果，直接返回同一事务SELECT得到的版本或显式绕过缓存。
动手实现
1. 用受影响行数确认修改，提交后再删除缓存
保存cache_write.py。MySQL没有此处的DML RETURNING，先用expected_version条件UPDATE并检查rowcount，再在同一持锁事务SELECT真实结果。UPDATE总是递增version，成功匹配必改变行。显式conn.commit成功后关闭连接，再调用Redis DEL；commit网络错误时结果可能不确定，不能盲重试。示例invalidated=False需由生产Outbox补偿，此文件未实现可靠投递。
文件：cache_write.py
from mysql_client import connect
from redis.exceptions import RedisError
from note_cache import key
from redis_client import r
def rename_note(note_id,title,expected_version):
conn=connect()
try:
with conn.cursor() as cur:
cur.execute("UPDATE notes SET title=%s,version=version+1,updated_at=UTC_TIMESTAMP(6) WHERE id=%s AND version=%s",
(title,note_id,expected_version))
if cur.rowcount!=1:
raise ValueError("not found or version conflict")
cur.execute("SELECT id,title,version FROM notes WHERE id=%s",(note_id,))
row=cur.fetchone()
conn.commit()
except BaseException:
conn.rollback()
raise
finally:
conn.close()
try:
r.delete(key(note_id))
invalidated=True
except RedisError:
invalidated=False
return {"id":int(row["id"]),"title":row["title"],"version":int(row["version"]),"invalidated":invalidated}
2. 验证成功修改后的首次回源
保存为 cache_invalidate.py 并运行。先预热旧值，再以当前版本更新。更新后键不存在，下一次 get_note 返回数据库的新标题与新版本。重复运行时从数据库读取当前版本，不假定总是 1。
文件：cache_invalidate.py
import os
from note_cache import get_note, key, load_note
from cache_write import rename_note
from redis_client import r
n = int(os.environ["NOTE_ID"])
before = load_note(n)
r.delete(key(n))
get_note(n)
print(rename_note(n, "cache03-new-title", before["version"]))
assert r.get(key(n)) is None
after, source = get_note(n)
print(source, after["title"], after["version"])
assert source == "database" and after["version"] == before["version"] + 1
3. 按确定时序重现旧值晚回填
本脚本将两个并发请求的步骤按顺序手工调度，结果不依赖线程调度或 sleep。只操作 NOTE_ID 专用笔记。A 先读取旧值并暂停，B 完成提交和失效，A 最后回填旧值；读取命中旧版本，数据库版本更高。
文件：cache_stale_refill.py
import os
from note_cache import load_note, get_note, encode, key
from cache_write import rename_note
from redis_client import r
n = int(os.environ["NOTE_ID"])
r.delete(key(n))
snapshot_a = load_note(n) # A: 已回源，尚未回填
written_b = rename_note(n, "cache03-race-new", snapshot_a["version"]) # B
r.set(key(n), encode(snapshot_a), ex=5) # A: 旧值晚到
cached, source = get_note(n)
fresh = load_note(n)
print("cached", source, cached["version"], "database", fresh["version"])
assert cached["version"] < fresh["version"]
r.delete(key(n)) # 清理本次故障注入
assert get_note(n)[0]["version"] == fresh["version"]
4. 制定可检查的失效策略
为 notes-service 的详情、列表、标签聚合分别列出受一次编辑影响的键。详情可精确删除；列表缓存需要版本化命名空间或短 TTL，不能用 KEYS 模糊删除整个实例。将下列记录保存为开发验收清单，后续 Outbox 模块实现重试。
文件：终端 / 当前课时
事件：note.updated，字段 note_id / committed_version / event_id
写入：业务 UPDATE 与 outbox INSERT 在同一个MySQL InnoDB事务（事件表DDL在部署阶段预先完成）
消费：按事件删除详情键及明确的派生键，成功后记录处理状态
重复：DEL 对不存在键仍可安全重放
失败：有限退避后进入待处理队列并告警，不丢弃失效事件
读取：写接口返回同事务SELECT得到并已commit的真实版本；强一致读取显式绕过缓存
观测：失效失败数、最老未处理事件年龄、回源率、旧版本样本
验收结果
- DEL 只出现在数据库事务成功退出后。
- 固定时序实验能得到缓存版本小于数据库版本，清理后恢复。
- 设计文档区分提交成功与失效成功，并给失效失败安排可靠补偿。
生产约束与故障演练
编辑提交成功后用户立即刷新，后台又可能存在较慢的并发读取。
实现决策
- 提交后删除详情缓存。
- 用确定性时序测试陈旧回填。
- 可靠失效事件与直接读写结果满足不同一致性需求。
- 事务中完成UPDATE与结果SELECT，commit成功后再失效；迁移DDL不混入业务事务。
隔离环境演练
仅对 NOTE_ID 练习笔记模拟旧读取在新提交之后回填，比较两个版本并清理故障值。
恢复验收
- 没有宣称 Cache-Aside 天然强一致。
- 失效故障不会伪装成数据库回滚。
- 派生缓存清单和重试责任明确。
常见错误
在异常里删缓存，用户仍看到旧值
原因：失效顺序没有跟随真实提交结果，或遗漏派生列表键。
处理：围绕事务提交定义事件，列出所有派生缓存；不要以请求函数是否执行完代替提交状态。
采用延迟双删后宣称缓存强一致
原因：延迟长短无法覆盖任意暂停、网络延迟和进程故障。
处理：把延迟删除视为减少窗口的措施；依据业务选择可接受陈旧、可靠事件或直接读真源。
扩展练习
数据库更新已提交但 Redis 断开时，接口能否返回“数据库操作已回滚”？应记录什么？
提示：数据库提交和 Redis 失效是两个独立结果。
参考答案：
不能报告回滚。commit已成功时应保留实际提交版本，记录待补偿事件及失败原因，使用同事务Outbox可靠重试缓存删除。若commit本身因网络中断返回错误，结果是不确定，需按幂等标识查询真源核对，不能假定没提交。
技术文档
04 阻止缓存穿透：输入校验与短期负缓存
对确定不存在的编号缓存短期空结果，并避免把故障或权限拒绝当作全局不存在。
本节目标
- 区分格式非法、不存在和无权限
- 为不存在结果设计明确编码和短 TTL
- 限制攻击者制造的负缓存键数量
必要概念
穿透是大量请求持续落到真源
查询从不存在的编号时，普通 Cache-Aside 每次都得到空结果而不回填，会重复访问数据库。输入范围校验先拦掉无意义请求，负缓存减少同一个合法但不存在编号的重复查询。
空结果必须有明确标记
Redis GET 返回 None 表示缓存键不存在，不等于业务记录不存在。负缓存使用 found=false；正缓存使用 found=true 和 note，读取时分别处理。数据库连接失败绝不能被编码为 found=false。
权限不能被共享缓存省略
记录缓存只是数据副本，不代替鉴权。多租户应用应先完成主体与租户校验，并把需要隔离的租户放进键名。单用户课程暂不引入租户字段；不能把某个用户的 403 缓存成所有用户都不存在。
动手实现
1. 实现正负结果独立编码
保存为 negative_cache.py。单独使用 v1:lookup 命名空间，不与第 02 课 JSON 结构混用。正结果 60 秒，确定无行 5 秒。类型校验拒绝 bool 作为整数编号；上限 2^63-1 对应数据库 bigint 的正数范围。
文件：negative_cache.py
import json
from redis_client import r
from note_cache import load_note,decode,encode
def lookup_key(note_id):
return f"notes:lab:v1:lookup:{{{note_id}}}"
def get_with_negative(note_id):
if type(note_id) is not int or not 1 <= note_id <= 2**63 - 1:
raise ValueError("invalid note id")
k = lookup_key(note_id)
raw = r.get(k)
if raw is not None:
try:
data = json.loads(raw)
if data["schema"] != 1 or type(data["found"]) is not bool:
raise ValueError("bad cache")
if data["found"] is False:
return None, "negative-cache"
note = data["note"]
if not isinstance(note, dict) or note.get("id") != note_id:
raise ValueError("bad note")
return decode(encode(note),note_id), "cache"
except (ValueError, KeyError, TypeError):
r.delete(k)
note = load_note(note_id) # 异常继续向外传递，不伪造不存在
found = note is not None
payload = {"schema": 1, "found": found, "note": note}
r.set(k, json.dumps(payload), ex=60 if found else 5)
return note, "database"
2. 选择确认不存在的MySQL编号并比较来源
只在独立练习库选MAX(id)+100000并确认在BIGINT正数范围。MAX这里只是产生确定不存在的实验输入，不用于获取刚插入记录的主键。成功只读事务也显式commit，异常rollback，最终close。
文件：negative_probe.py
from mysql_client import connect
from negative_cache import get_with_negative,lookup_key
from redis_client import r
conn=connect()
try:
with conn.cursor() as cur:
cur.execute("SELECT COALESCE(MAX(id),0)+100000 AS missing_id FROM notes")
missing_id=int(cur.fetchone()["missing_id"])
conn.commit()
finally:
conn.rollback()
conn.close()
assert 1<=missing_id<=2**63-1
r.delete(lookup_key(missing_id))
print(get_with_negative(missing_id))
print(get_with_negative(missing_id))
print("ttl",r.ttl(lookup_key(missing_id)))
r.delete(lookup_key(missing_id))
3. 让非法编号在访问存储之前失败
保存并运行。bad 值都会触发 ValueError；后续 API 接入时由参数模型先执行同等校验。单个非法值的验证不要发起 Redis 或 MySQL 请求。
文件：negative_invalid.py
from negative_cache import get_with_negative
for bad in [0, -1, True, "12", 2**63]:
try:
get_with_negative(bad)
except ValueError as exc:
print(repr(bad), str(exc))
else:
raise AssertionError("invalid id reached storage")
4. 把创建与更新失效覆盖到负缓存
将 lookup_key 对应键加入提交后失效清单。新建时数据库返回 id 后再删除同一编号的负缓存；更新与删除时同时失效详情和 lookup 两种命名空间。下面函数只演示失效动作，调用必须放在业务事务提交后。
文件：cache_invalidate_keys.py
from redis_client import r
from note_cache import key
from negative_cache import lookup_key
def invalidate_note_caches(note_id):
return r.delete(key(note_id), lookup_key(note_id))
# 调用位置：数据库提交成功后，或可靠 Outbox 消费函数中。
# 不在这里捕获错误后无声丢弃。
验收结果
- 无行结果只回源一次，短期第二次返回 negative-cache。
- 非法编号在存储访问之前被拒绝。
- 新建、修改和删除的失效清单包含负缓存键，数据库错误不写入负缓存。
生产约束与故障演练
重复访问不存在笔记的请求增加，数据库空查询消耗开始明显。
实现决策
- 格式校验先于缓存读取。
- 只缓存确定的无行结果，负 TTL 短于正常缓存。
- 负缓存结合入口限流与容量监控。
隔离环境演练
对确定不存在的练习编号连续读取，再提交非法编号，比较来源和存储请求。
恢复验收
- 不存在与故障分开。
- 没有以负缓存绕过鉴权。
- 新增路径失效负缓存。
常见错误
新建笔记后短时间仍显示不存在
原因：创建前留下负缓存，创建后没有失效。
处理：用数据库返回的真实 id 删除对应负缓存，并设置较短负 TTL。
缓存里出现海量随机不存在编号
原因：攻击者每次换编号，短 TTL 也无法阻止瞬时键膨胀。
处理：在入口鉴权和限流，限制参数与访问范围；观察负缓存创建率和内存，不依赖负缓存单独抗攻击。
扩展练习
如果增加布隆过滤器，返回“可能存在”和“确定不在过滤器”各意味着什么？
提示：过滤器有误判，且构建及更新是否完整决定能否用来拒绝查询。
参考答案：
可能存在仍需查缓存或数据库。确定不在过滤器只有在过滤器完整且新增写入同步维护时才可拒绝；过滤器延迟会误拒绝新记录，因此需要回源或版本化切换策略。它不能代替权限检查与数据库真源。
技术文档
05 抑制热点击穿：带令牌的单飞回填
让同一热点键过期时只有少量请求回源，使用原子令牌检查防止旧持有者误删新锁。
本节目标
- 用 SET NX PX 合并同一键的回源
- 通过 Lua 原子校验令牌和释放
- 给竞争者等待与锁过期设计明确结果
必要概念
单飞用于压低重复工作
同一个热点缓存失效时，许多请求同时回源会压垮数据库。短期填充锁让一个请求重建，其他请求短暂等待缓存或收到忙响应。它服务于缓存效率，不承诺跨故障条件下业务副作用严格只执行一次。
释放时检查所有者令牌
每次竞争生成随机 token，并用 SET lock token NX PX 同时设置所有权和租期。释放必须在一段 Lua 中比较 token 再删除；先 GET 后 DEL 存在两条命令之间被新持有者接手的窗口。Redis 7 示例使用 Lua，不使用较新版本的条件删除命令。
过期意味着可能出现另一位持有者
进程暂停或回源过慢会让锁先过期。旧持有者恢复后不能删除或发布覆盖新持有者结果。下面脚本把校验、回填与释放放在一次 Lua 中；数据库并发更新仍有上一课的陈旧回填问题，单飞并不自动解决缓存一致性。
动手实现
1. 保存安全释放脚本
保存为 cache_singleflight.py，后续步骤继续向此文件追加。键名使用同一个 {note_id} 标签，便于将来多键脚本要求同槽；当前未部署 Redis Cluster。脚本只有固定数量短操作，不在 Redis 内做长循环或外部访问。
文件：cache_singleflight.py
import secrets
import time
from redis_client import r
from note_cache import key, load_note, encode, decode
class CacheBusy(RuntimeError):
pass
release = r.register_script("""
if redis.call('GET', KEYS[1]) == ARGV[1] then
return redis.call('DEL', KEYS[1])
end
return 0
""")
publish = r.register_script("""
if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end
redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3])
redis.call('DEL', KEYS[1])
return 1
""")
2. 实现有限等待与二次读取
追加到 cache_singleflight.py。获得锁后再读一次缓存，避免等待途中别人已经回填。竞争失败最多等 0.3 秒，每 20 毫秒看一次结果；耗尽后 CacheBusy，应映射为受控 503 或按策略返回允许的旧值。不能让所有竞争者超时后同时无限回源。
文件：cache_singleflight.py
def singleflight_note(note_id, loader=load_note):
cache_key, lock_key = key(note_id), key(note_id) + ":fill"
raw = r.get(cache_key)
if raw is not None:
return decode(raw, note_id)
token = secrets.token_urlsafe(24)
if r.set(lock_key, token, nx=True, px=2000):
try:
raw = r.get(cache_key)
if raw is not None:
return decode(raw, note_id)
note = loader(note_id)
if note is None:
return None
stored = publish(keys=[lock_key, cache_key], args=[token, encode(note), 60000])
if not stored:
raise CacheBusy("fill lease expired")
return note
finally:
release(keys=[lock_key], args=[token])
deadline = time.monotonic() + 0.3
while time.monotonic() < deadline:
time.sleep(0.02)
raw = r.get(cache_key)
if raw is not None:
return decode(raw, note_id)
raise CacheBusy("cache fill still running")
3. 并发验证回源次数
保存为 cache_singleflight_probe.py。8 个线程同时读取同一笔记，模拟回源增加 80 毫秒。正常本机负载下 calls 应为 1；若发生 Redis 超时或调度超过等待期限，应把失败视为实际故障记录，不删除断言掩盖问题。专用记录必须存在，运行前清理详情和填充锁。
文件：cache_singleflight_probe.py
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from note_cache import key, load_note
from redis_client import r
from cache_singleflight import singleflight_note
n = int(os.environ["NOTE_ID"])
r.delete(key(n), key(n) + ":fill")
guard = threading.Lock()
calls = 0
def counted_loader(note_id):
global calls
with guard:
calls += 1
time.sleep(0.08)
return load_note(note_id)
with ThreadPoolExecutor(max_workers=8) as workers:
rows = list(workers.map(lambda _: singleflight_note(n, counted_loader), range(8)))
assert all(row is not None and row["id"] == n for row in rows)
assert calls == 1
print("database loads", calls)
4. 验证旧令牌不能释放新锁
保存为 cache_lock_token.py。只对本课 token-probe 键模拟替换所有者，原令牌释放返回 0，键仍为 new-owner。这个实验直接设置新 token 以固定复现所有权变化；真实场景通常由原锁过期后竞争产生。
文件：cache_lock_token.py
from redis_client import r
from cache_singleflight import release
k = "notes:lab:token-probe"
r.set(k, "old-owner", px=2000)
r.set(k, "new-owner", px=2000)
assert release(keys=[k], args=["old-owner"]) == 0
assert r.get(k) == "new-owner"
assert release(keys=[k], args=["new-owner"]) == 1
print("ownership check passed")
验收结果
- 正常热点并发实验只回源一次，其余读取缓存结果。
- 旧 token 不能删除新 token 的锁。
- 等待、租期、Redis 超时各有上限，锁过期后不发布旧持有者结果。
生产约束与故障演练
热门笔记缓存到期后突然出现大量相同详情请求。
实现决策
- 先用进程内合并或单实例填充锁降低重复回源。
- 令牌检查与发布、释放原子执行。
- 竞争失败有界等待，不把所有失败者直接推向数据库。
隔离环境演练
在本机令牌键替换所有者，再延长 loader 到超过租期，检查旧持有者行为。
恢复验收
- 旧持有者不能删新锁。
- 数据库回源次数有度量。
- 不把缓存锁当作生产业务唯一性保证。
常见错误
finally 中 DEL 删除了别人的锁
原因：原锁过期后另一请求获得锁，旧请求无条件删除。
处理：使用 Lua 比较 token 并原子删除，不能把 GET 与 DEL 分开发。
把缓存锁用于扣款并宣称绝不重复
原因：租期过期、进程暂停或异步复制故障会破坏单持有者假设。
处理：业务正确性使用数据库约束、幂等键或具备 fencing 校验的资源；缓存锁只减少重复回源。
扩展练习
把 loader 延迟改为 2.2 秒，预测原持有者和竞争者结果，并说明生产应如何处理。
提示：填充租期 2 秒，竞争者等待 0.3 秒；二者是不同预算。
参考答案：
竞争者先收到 CacheBusy；原持有者回源完成后 publish 因 token 已过期返回 0，也抛出 CacheBusy。应该限制回源耗时、选择合适租期并监控失败；若需要续租必须验证所有者且限定总时间，不能无限延长。
技术文档
06 应对缓存雪崩：TTL 抖动、受控预热与降级
分散批量过期时间，限制冷启动回源并发，在 Redis 失效时保护 MySQL。
本节目标
- 给 TTL 加有界随机偏移
- 用限速小批次预热热点数据
- 让缓存降级服从数据库并发预算
必要概念
雪崩影响一批键或整个实例
同一时刻大量键到期，或 Redis 整体不可用，会把原本被缓存吸收的流量推向数据库。TTL 抖动分散到期时刻，不能解决实例故障；需要回源限流和服务端负载保护。
降级也有容量上限
Redis故障不代表MySQL能接住全部流量。回源需按进程设置并发名额、池等待、驱动读写超时、InnoDB锁等待和SELECT执行预算；超量返回明确暂不可用。即使数据库默认RR，应用连接也显式RC并结束每次只读事务，避免复用连接时长期保留旧快照。
预热从热点开始
预热最常访问的有限集合，并控制批次大小、并发和速率。全库一起加载会抢占生产数据库资源和缓存容量。旧值兜底只适用于允许陈旧的字段，权限、删除状态和敏感信息不能随意继续返回。
动手实现
1. 观察有界 TTL 抖动分布
保存为 cache_jitter.py。生成 100 个专用小键，TTL 均在 45 到 75 秒，真实设置后可能因执行消耗略减 1 秒。只清理本脚本创建的键，不扫描并删除其他缓存。
文件：cache_jitter.py
import random
from collections import Counter
from redis_client import r
keys = [f"notes:lab:jitter:{i}" for i in range(100)]
ttls = [60 + random.randint(-15, 15) for _ in keys]
with r.pipeline(transaction=False) as pipe:
for k, ttl in zip(keys, ttls):
pipe.set(k, "sample", ex=ttl)
pipe.execute()
print("min/max", min(ttls), max(ttls))
print("buckets", sorted(Counter(ttls).items()))
r.delete(*keys)
2. 用MySQL有限读取预热最近20条
保存cache_warm.py。先在短事务读取最多20个编号，commit并close后再逐个回源填充。每次load_note都重新打开短RC读取，删除期间得到None则跳过；不是一致性全库导出。顺序预热、每条50毫秒与TTL抖动限制本机负载，生产使用池及实际热点统计。
文件：cache_warm.py
import random
import time
from mysql_client import connect
from note_cache import key,load_note,encode
from redis_client import r
conn=connect()
try:
with conn.cursor() as cur:
cur.execute("SELECT id FROM notes ORDER BY id DESC LIMIT 20")
ids=[int(row["id"]) for row in cur.fetchall()]
conn.commit()
except BaseException:
conn.rollback()
raise
finally:
conn.close()
for note_id in ids:
note=load_note(note_id)
if note is not None:
r.set(key(note_id),encode(note),ex=60+random.randint(-15,15))
time.sleep(0.05)
print("warmed",len(ids))
3. 实现有并发上限的回源降级
保存为 cache_fallback.py。该函数直接实现读路径，不调用可能已经回源的 get_note 后再次回源，避免重复数据库查询。Redis GET/SET 失败可以绕过，MySQL 异常仍交给调用者。Semaphore 只约束本进程；多进程预算需要总量规划。同步代码不能直接阻塞 async 路由。
文件：cache_fallback.py
import random
import threading
from redis.exceptions import RedisError
from redis_client import r
from note_cache import key, load_note, encode, decode
db_slots = threading.BoundedSemaphore(4)
class SourceBusy(RuntimeError):
pass
def get_available(note_id, loader=load_note):
try:
raw = r.get(key(note_id))
if raw is not None:
try:
return decode(raw, note_id), "cache"
except (ValueError, KeyError, TypeError):
pass
except RedisError:
pass # 生产记录 redis_read_error 计数，不记录密码
if not db_slots.acquire(timeout=0.05):
raise SourceBusy("database fallback capacity exhausted")
try:
note = loader(note_id)
finally:
db_slots.release()
if note is not None:
try:
r.set(key(note_id), encode(note), ex=60 + random.randint(-15, 15))
except RedisError:
pass # 生产记录 redis_fill_error 计数
return note, "database"
4. 停掉本机缓存并验证仍可读取真源
先在终端执行 docker stop notes-redis，仅影响本课专用实例；运行下面脚本，预期来源 database。完成后 docker start notes-redis 并再运行两次，后续可恢复 cache。不要在服务正在高负载时无保护地执行全局故障演练。
文件：cache_outage_probe.py
import os
from cache_fallback import get_available
note, source = get_available(int(os.environ["NOTE_ID"]))
assert note is not None
print(source, note["id"], note["version"])
# Shell 完成后恢复：docker start notes-redis
# 本课 Redis 未开启持久化，重启后的缓存需要重新构建。
验收结果
- TTL 分布有上下界，批量键没有全部同刻过期。
- 预热数量和速率有限，未把全库一次性加载。
- 专用 Redis 停止时仍读到 MySQL 数据，回源超出预算时返回受控错误。
生产约束与故障演练
发布或 Redis 重启后大量缓存同时为空，数据库必须保持可用。
实现决策
- 到期抖动、限速预热与故障降级分别处理不同触发条件。
- 数据库回源有硬并发预算。
- 允许陈旧数据的范围由业务定义。
- 回源同步驱动为PyMySQL；异步API调用改用对应asyncmy接入，不在事件循环中直接执行同步SQL。
隔离环境演练
仅停止 notes-redis，使用有限并发读取并测量回源峰值，随后恢复实例。
恢复验收
- 不会无限回源或无限等待。
- 数据库错误仍可见。
- 故障恢复后命中率逐步恢复而非立即全量预热。
常见错误
缓存断开后数据库连接数瞬间暴涨
原因：降级没有并发限制，重试与每次新建连接叠加。
处理：按进程规划回源预算，使用池和短等待，超过容量快速拒绝；避免多层重试。
抖动后出现零或负 TTL
原因：随机偏移超过基础期限。
处理：明确最小 TTL 并测试范围，例如 max(1,base+jitter)，业务过期上限也要符合要求。
扩展练习
在 Redis 停止时用 12 个线程和每次 0.2 秒的 loader 调用 get_available，预测 SourceBusy 的原因。
提示：本进程只有 4 个回源名额，等待最多 0.05 秒。
参考答案：
前 4 个请求可占用回源名额，其余请求可能在名额释放前超时并得到 SourceBusy。实际数量由调度决定，验收应看峰值回源并发不超过 4、等待有界、恢复后正常，而非硬编码某个固定失败数。
技术文档
07 使用 Pipeline、WATCH 与 Lua 保证正确的原子操作
比较批量发送、乐观事务和 Lua，在 Redis 7 中实现带过期窗口的原子计数限流。
本节目标
- 区别流水线减少往返与事务原子执行
- 处理 WATCH 发现的并发修改
- 用短 Lua 脚本组合 INCR 和到期设置
必要概念
Pipeline 首先减少网络往返
transaction=False 的 pipeline 批量发送命令，不让多条业务操作自动变成一个不可分割动作。redis-py pipeline 默认启用事务，教学中明确写参数以免混淆。大批次会占内存和延长等待，应限制规模。
WATCH 是提交前的乐观检查
WATCH 键后读取当前值，MULTI 与 EXEC 提交修改；被其他客户端改动时 redis-py 抛 WatchError，应用重新读取并有限重试。Redis 事务与 MySQL 不同，执行阶段某条命令出错不会撤销已经成功的其他命令。
Lua 组合命令时先验证输入
短 Lua 脚本在执行期间不被其他命令穿插。原子执行不代表发生运行时错误会回滚此前的写入。参数与数据类型应在变更前校验，脚本必须短，不能用长循环阻塞所有客户端。
动手实现
1. 批量发送并理解返回位置
保存为 redis_pipeline.py。transaction=False 只减少往返，返回列表按命令顺序排列。专用键全部带 TTL，并在验收后显式删除。若一次批量中途网络断开，不能仅凭客户端异常判断所有命令都没执行。
文件：redis_pipeline.py
from redis_client import r
keys = [f"notes:lab:pipeline:{i}" for i in range(3)]
with r.pipeline(transaction=False) as pipe:
for i, k in enumerate(keys):
pipe.set(k, str(i), ex=60)
for k in keys:
pipe.get(k)
result = pipe.execute()
print(result)
assert result == [True, True, True, "0", "1", "2"]
r.delete(*keys)
2. 制造 WATCH 冲突并重读
保存为 redis_watch.py。第一次 watch 后，使用客户端的另一次命令修改该键，确保 EXEC 抛 WatchError；第二次重新 watch 和读取，应从值 10 增加到 11。最多尝试 3 次，示例没有无限循环。
文件：redis_watch.py
from redis import WatchError
from redis_client import r
k = "notes:lab:watch-counter"
r.set(k, 0, ex=60)
for attempt in range(3):
with r.pipeline() as pipe:
try:
pipe.watch(k)
current = int(pipe.get(k) or 0)
if attempt == 0:
r.set(k, 10, ex=60) # 另一次命令制造冲突
pipe.multi()
pipe.set(k, current + 1, ex=60)
pipe.execute()
break
except WatchError:
print("conflict, retry from read")
else:
raise RuntimeError("retry budget exhausted")
assert r.get(k) == "11"
r.delete(k)
3. 实现原子计数与固定期限
保存为 redis_limiter.py。脚本验证限制与窗口后 INCR，首次计数同时 PEXPIRE，避免 INCR 成功但 EXPIRE 未发出的永久计数器。窗口从首次请求开始，不是精确滑动窗口。网络超时重试可能再次计数，业务必须允许保守限流或使用独立请求幂等方案。
文件：redis_limiter.py
from redis_client import r
limit_script = r.register_script("""
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
if not limit or limit < 1 or limit % 1 ~= 0 then
return redis.error_reply('invalid limit')
end
if not window or window < 1 or window % 1 ~= 0 then
return redis.error_reply('invalid window')
end
local count = redis.call('INCR', KEYS[1])
if count == 1 then redis.call('PEXPIRE', KEYS[1], window) end
local allowed = 0
if count <= limit then allowed = 1 end
return {allowed, count, redis.call('PTTL', KEYS[1])}
""")
def allow_request(subject):
# subject 必须是鉴权后得到的稳定主体 ID，不能是任意超长输入。
k = f"notes:lab:limit:{{{subject}}}"
allowed, count, ttl_ms = limit_script(keys=[k], args=[3, 1000])
return bool(allowed), count, ttl_ms
4. 验证窗口内拒绝和到期恢复
保存为 redis_limiter_probe.py。前 3 次允许，第 4 次拒绝且返回正的剩余毫秒；约 1.1 秒后新窗口允许。生产映射 429，并根据 TTL 向上取整计算 Retry-After；权限鉴定接口与普通缓存故障时的放行策略必须分别设计。
文件：redis_limiter_probe.py
import time
from redis_client import r
from redis_limiter import allow_request
k = "notes:lab:limit:{probe-user}"
r.delete(k)
results = [allow_request("probe-user") for _ in range(4)]
print(results)
assert [x[0] for x in results] == [True, True, True, False]
assert results[-1][2] > 0
time.sleep(1.1)
assert allow_request("probe-user")[0] is True
r.delete(k)
验收结果
- Pipeline 返回顺序与命令顺序一致，批量大小有限。
- WATCH 实验先失败再从新值重试，不覆盖外部修改。
- 限流前三次允许、第四次拒绝、窗口到期恢复，计数键存在时带 TTL。
生产约束与故障演练
API 需要在多进程之间共享请求计数，并避免计数和过期配置分离。
实现决策
- 流水线、事务和 Lua 各用于明确的操作边界。
- 脚本短且输入受限。
- 限流故障策略不沿用普通数据缓存的无条件放行。
隔离环境演练
在专用键制造 WATCH 冲突与窗口超额请求，验证有限重试和原子期限。
恢复验收
- 没有永久计数键。
- 事务错误不被错误描述为自动回滚。
- 超时重试与边界突发的业务含义明确。
常见错误
误把 transaction=False 的 Pipeline 当成原子业务操作
原因：减少网络往返与命令不可交错是不同保证。
处理：需要检查并更新用 WATCH，需要短组合操作用 Lua；明确事务选项。
INCR 后进程崩溃留下永不过期限流键
原因：EXPIRE 是另一条命令，没来得及发出。
处理：把首次计数和到期设置放进短脚本；对专用命名空间监控 TTL=-1 异常。
扩展练习
窗口边界附近为什么可能在很短时间内通过接近两倍限制的请求？
提示：上一窗口末尾和下一窗口开头分别拥有各自额度。
参考答案：
固定窗口允许边界突发：旧窗口末尾用完额度，过期后新窗口立即重新计数。需要更平滑限制时可选择滑动窗口或令牌桶，评估精度、状态量和脚本成本；不能把本例当作精确的任意 1 秒最多 3 次。
技术文档
08 控制缓存容量与淘汰，隔离任务 Broker
在专用容量实例制造淘汰，观察命中率和内存，再区分可丢缓存与不可随意丢弃的队列数据。
本节目标
- 读取内存、淘汰和命中指标
- 根据数据用途选择 allkeys-lru 或 noeviction 等策略
- 解释为什么逻辑数据库不能替代实例隔离
必要概念
maxmemory 不是进程总内存上限
Redis 的 maxmemory 控制参与淘汰计算的内存范围，进程 RSS、复制缓冲、持久化重写等还需要额外预算。容器内存上限应留余量，不能把 maxmemory 与容器限制设成完全相同。
淘汰与到期不同
到期按 TTL 删除，淘汰在内存压力下依据策略移除键。allkeys-lru 可从所有键中近似选择较久未使用的键；noeviction 在不足时拒绝某些需要新增内存的写命令。没有任何策略保证缓存永不丢失。
缓存实例不承载任务可靠性
任务 Broker、结果存储、锁与缓存可能要求不同持久化、淘汰和资源策略。逻辑 DB 0 与 DB 1 仍共享同一进程内存、淘汰配置和故障域。队列数据应按任务系统要求使用独立实例，noeviction 也只是拒写，不会自动让队列可靠。
动手实现
1. 启动单独容量实验实例
以下 Linux Bash 只创建 notes-redis-capacity，宿主机 6381，与业务练习缓存 6380 分开。读取现有本机配置并只修改 maxmemory。不要对共享 Redis 执行 CONFIG SET 作为临时实验。镜像实例仍使用已有 REDIS_PASSWORD。
文件：终端 / 当前课时
python3 - <<'PY'
from pathlib import Path
source = Path('redis-lab/redis.conf').read_text()
Path('redis-lab/capacity.conf').write_text(source.replace('maxmemory 128mb', 'maxmemory 8mb'))
PY
docker run --name notes-redis-capacity -d -p 127.0.0.1:6381:6379 \
-v "$PWD/redis-lab/capacity.conf:/usr/local/etc/redis/redis.conf:ro" \
redis:7 redis-server /usr/local/etc/redis/redis.conf
2. 有限写入制造容量压力
保存为 redis_capacity.py。最多写 2,000 个约 8KB 字符串，带 10 分钟 TTL；请求数和内存实验都有上限。运行前核对端口 6381。记录写入前后 evicted_keys 差值，不能用实例启动以来累计值直接推算本次淘汰。
文件：redis_capacity.py
import os
import redis
client = redis.Redis(host="127.0.0.1", port=6381, password=os.environ["REDIS_PASSWORD"],
decode_responses=True, socket_timeout=1, socket_connect_timeout=1)
before = client.info("stats")
for i in range(2000):
client.set(f"notes:lab:capacity:{i}", "x" * 8192, ex=600)
after = client.info("stats")
memory = client.info("memory")
print("evicted delta", after["evicted_keys"] - before["evicted_keys"])
print("used / rss / max", memory["used_memory"], memory["used_memory_rss"], memory["maxmemory"])
print("remaining keys", client.dbsize())
print("policy", client.config_get("maxmemory-policy"))
client.close()
3. 区分命中率与淘汰率
在容量实例读取 2,000 个明确键，统计本次命中和未命中。下面本地计数是这批请求的真实比例；INFO 的 keyspace_hits/misses 是实例全局累计值，做监控时用区间增量。不要用 KEYS * 在大实例全量扫描。
文件：redis_capacity_hits.py
import os
import redis
client = redis.Redis(host="127.0.0.1", port=6381, password=os.environ["REDIS_PASSWORD"],
decode_responses=True, socket_timeout=1)
hits = sum(client.get(f"notes:lab:capacity:{i}") is not None for i in range(2000))
print("hits", hits, "misses", 2000-hits, "ratio", hits/2000)
stats = client.info("stats")
print({k: stats[k] for k in ["keyspace_hits", "keyspace_misses", "evicted_keys", "expired_keys"]})
client.close()
4. 形成缓存与 Broker 的独立配置
把下列配置记录加入部署设计。容量实验完成后执行 docker stop notes-redis-capacity，保留环境便于复查，不触碰 6380。若要演练 noeviction，应复制配置另建独立实例，以收到内存不足写入错误为预期，并验证任务生产端如何处理拒写。
文件：终端 / 当前课时
普通笔记缓存：独立 cache 实例；明确 maxmemory；可用 allkeys-lru/LFU；值可从 MySQL 重建
任务 Broker：独立实例；按任务框架要求设置持久化与 noeviction；拒写必须使投递失败可见
任务结果：独立 TTL 与保留期；持久业务状态放 MySQL
容量预算：容器限制 > maxmemory，并留复制、AOF 重写、碎片等余量
告警：used_memory/maxmemory、RSS、evicted_keys 增量、命中率、写入拒绝数
检查：DB 0 与 DB 1 仅键空间区分，不提供内存、CPU 或故障隔离
验收结果
- 专用容量实例 evicted_keys 增长，键数量小于写入总数。
- 命中率从本次读取统计，INFO 累计指标用增量解释。
- 设计明确缓存与 Broker 分实例、不同持久化和拒写处理。
生产约束与故障演练
缓存规模增长且后台任务开始共用 Redis，需要避免缓存压力导致任务异常。
实现决策
- 实验容量实例独立于其他课程环境。
- 按进程 RSS 与服务负载规划余量。
- 可丢数据与可靠任务隔离。
隔离环境演练
在 6381 专用实例有限写入大值，观察淘汰与命中变化，完成后停止该实例。
恢复验收
- 未改动 6380 缓存策略。
- 队列拒写和任务恢复有处理路径。
- 告警使用区间指标与业务影响而非单一内存数值。
常见错误
Redis 尚未到 maxmemory，容器已经 OOM
原因：忽略 RSS、缓冲和持久化工作期间额外内存。
处理：对照容器指标和 used_memory_rss，留容量余量并压测重写与复制场景。
任务突然丢失，但 Redis 没报错
原因：队列与普通缓存共用允许淘汰所有键的实例。
处理：按 Broker 要求迁移到独立实例并禁止随意淘汰，同时完善任务确认、重投和持久业务状态。
扩展练习
为什么把 Celery Broker 放 DB 1、缓存放 DB 0 仍可能互相影响？
提示：想清楚 maxmemory、CPU、进程退出和持久化属于哪一层。
参考答案：
逻辑库共享同一 Redis 进程和服务器级配置，缓存写入可触发整体内存压力，CPU 长命令和实例故障也影响队列。需要独立实例与资源预算；不同 DB 编号只能减少键名碰撞。
技术文档
09 配置连接池、超时与熔断并记录缓存指标
让 Redis 故障快速暴露，短期跳过持续失败的访问，并在恢复后用探测重新启用缓存。
本节目标
- 分别设置连接、读写和池等待超时
- 实现进程内有限状态熔断示例
- 记录命中、回源、错误与等待指标
必要概念
超时需要放进请求总预算
连接超时、socket 超时、池等待和重试时间会叠加。API 总预算只有一秒时，不能给每层都设置一秒再重试三遍。写命令超时可能已经执行，重试 INCR、队列入队等操作需要额外语义设计。
熔断短期跳过已知失败依赖
连续 Redis 连接错误达到阈值后短暂打开熔断；冷却后只允许一个探测请求，其余请求继续走受控回源。恢复探测成功才关闭。下面是线程安全的进程内教学实现，不是跨所有进程同步的全局开关。
命中率不是唯一目标
高命中率也可能返回陈旧数据，低命中率也可能是新内容正常访问。至少同时观察缓存调用延迟、连接错误、回源并发、数据库延迟、失效失败与版本新鲜度。计数标签不能使用每个 note_id，否则监控基数失控。
动手实现
1. 显式配置池与关闭顺序
保存为 redis_pool_client.py。BlockingConnectionPool 让借不到连接时最多等 50 毫秒；connect/socket 各 100 毫秒，Retry(...,0) 禁止本层自动重试，避免示例将预算偷偷翻倍。同步客户端供同步代码使用，async 路由选择 redis.asyncio 并 await。
文件：redis_pool_client.py
import os
import redis
from redis.backoff import NoBackoff
from redis.retry import Retry
pool = redis.BlockingConnectionPool(
host=os.environ.get("REDIS_HOST", "127.0.0.1"),
port=int(os.environ.get("REDIS_PORT", "6380")),
password=os.environ["REDIS_PASSWORD"],
decode_responses=True, max_connections=8, timeout=0.05,
socket_connect_timeout=0.1, socket_timeout=0.1,
health_check_interval=30, retry=Retry(NoBackoff(), 0),
)
client = redis.Redis(connection_pool=pool)
if __name__ == "__main__":
try:
print(client.ping())
finally:
client.close()
pool.disconnect()
2. 实现线程安全的短期熔断器
保存为 redis_breaker.py。只包裹 Redis 操作；业务 SQL 和 JSON 解析错误不能自动判作 Redis 断网。AUTH 等配置错误应报警修配置，不用无限探测掩盖。调用方对返回的 bypass 标志进入第 06 课的有界回源流程。
文件：redis_breaker.py
import threading
import time
from redis.exceptions import AuthenticationError, ConnectionError, TimeoutError
class RedisBreaker:
def __init__(self, threshold=3, cooldown=2):
self.threshold, self.cooldown = threshold, cooldown
self.failures = 0
self.open_until = 0.0
self.probing = False
self.generation = 0
self.lock = threading.Lock()
def call(self, operation):
now = time.monotonic()
with self.lock:
if self.open_until > now or self.probing:
return None, "bypass"
if self.open_until:
self.probing = True
self.generation += 1
generation = self.generation
try:
value = operation()
except AuthenticationError:
with self.lock:
if generation == self.generation:
self.probing = False
raise # 配置错误直接上报，不伪装成普通缓存未命中
except (ConnectionError, TimeoutError):
with self.lock:
if generation == self.generation:
self.failures += 1
if self.probing or self.failures >= self.threshold:
self.open_until = time.monotonic() + self.cooldown
self.generation += 1
self.probing = False
return None, "redis-error"
except BaseException:
with self.lock:
if generation == self.generation:
self.probing = False
raise
with self.lock:
if generation == self.generation:
self.failures, self.open_until, self.probing = 0, 0.0, False
return value, "redis"
3. 不用停服务验证状态转移
保存为 redis_breaker_probe.py。注入固定抛错函数，前 3 次真实失败，第 4 次直接 bypass；冷却后成功探测，下一次正常执行。此脚本先验证顺序行为；实现中的 generation 用于丢弃旧一代在途请求对新状态的覆盖，并发测试还需验证半开阶段只启动一个探测。
文件：redis_breaker_probe.py
import time
from redis.exceptions import ConnectionError
from redis_breaker import RedisBreaker
breaker = RedisBreaker(threshold=3, cooldown=0.1)
calls = 0
def broken():
global calls
calls += 1
raise ConnectionError("simulated unavailable")
for _ in range(4):
print(breaker.call(broken))
assert calls == 3
time.sleep(0.11)
assert breaker.call(lambda: "PONG") == ("PONG", "redis")
assert breaker.call(lambda: "cached") == ("cached", "redis")
print("recovered")
4. 记录实际调用状态与延迟
保存为 cache_metrics_probe.py。故障实验只停止 notes-redis 后运行，再重启后运行。下面状态为教学输出，生产导出聚合 counter/histogram，不记录密码、完整值或每条笔记作为标签。Redis miss 是 value=None 且 status=redis，不能与 bypass 合并成命中。
文件：cache_metrics_probe.py
import os
import time
from note_cache import key
from redis_pool_client import client, pool
from redis_breaker import RedisBreaker
breaker = RedisBreaker()
try:
for _ in range(5):
started = time.perf_counter()
value, status = breaker.call(lambda: client.get(key(int(os.environ["NOTE_ID"]))))
hit = status == "redis" and value is not None
print({"status":status,"hit":hit,"duration_ms":round((time.perf_counter()-started)*1000,2)})
finally:
client.close()
pool.disconnect()
验收结果
- 池等待、连接、socket 超时与重试次数均显式设置。
- 失败阈值、bypass、冷却探测与恢复能通过注入实验验证。
- 指标区分命中、未命中、Redis 错误和主动绕过，不记录高基数或敏感数据。
生产约束与故障演练
Redis 网络故障时，API 需要在可接受延迟内切换到受控回源，并在恢复后重新命中。
实现决策
- 超时预算以完整请求为单位计算。
- 进程内熔断只负责快速跳过 Redis，回源仍受容量限制。
- 通过状态代次阻止在途旧响应覆盖新状态，并对半开探测做并发验收。
隔离环境演练
注入连续连接错误验证熔断转移，再用专用实例停止恢复观察实际延迟和来源。
恢复验收
- 故障延迟有上限。
- 没有把所有 Redis 错误当成普通 miss。
- 连接池在进程退出时关闭。
常见错误
Redis 已故障，但每个请求仍等很久
原因：多层超时和自动重试叠加，没有总预算。
处理：画出一次请求的所有等待预算，限制重试所在层；以实测故障延迟验收。
一个并发成功请求把刚打开的熔断立即关闭
原因：多个已在途调用同时回写熔断状态，若删除 generation 检查就会出现旧请求覆盖新状态。
处理：保留 generation 校验：打开熔断或进入半开时推进代次，旧代次结果不改变新状态；补充并发探测测试或使用已有成熟组件。
扩展练习
为什么缓存 GET 超时可以考虑回源，而任务入队超时不能直接断言“任务没入队”？
提示：响应丢失不等于服务器没执行命令。
参考答案：
GET 是读取，失败后可按预算回源；入队命令可能已执行但响应丢失，直接再次入队可能重复。使用任务 ID、幂等消费、投递记录与状态核对，重试策略按操作语义设计。
技术文档
10 演练重启、复制与持久化恢复
对比可重建缓存与开启 AOF 的实验实例，检查副本状态，并建立故障后的业务验收顺序。
本节目标
- 验证无持久化缓存重启后的回源
- 在隔离实例验证 AOF 重启恢复
- 理解异步复制不等于零数据丢失或自动高可用
必要概念
RDB 与 AOF 保存不同形式
RDB 保存某个时点的快照；AOF 记录写操作并按 fsync 策略落盘。appendfsync everysec 在故障下仍有近期数据丢失窗口。正常重启看到值还在，只证明该次正常路径，不能据此宣称断电零丢失。
复制不是自动故障转移
Redis 主从默认异步复制，主节点确认写入时副本可能尚未收到。WAIT 可等待副本确认复制，仍不把系统变成强一致存储。生产自动故障转移由 Sentinel、Cluster 或受管平台等承担，部署拓扑与客户端发现都需独立验证。
恢复验收从真源开始
先确认MySQL 8.4业务数据完整、InnoDB可读写，再检查Redis角色、复制与受控回源。Redis AOF、RDB与MySQL的redo、binlog是不同日志，不能互相替代。缓存可丢时可从MySQL重建；任务Broker还需验收持久Outbox、未确认任务与幂等消费，不用缓存重启结果证明任务可靠性。
动手实现
1. 确认无持久化缓存重启后可回源
使用第 02 课 NOTE_ID 预热后，记录缓存存在。执行 docker restart notes-redis，仅重启专用本机实例，然后再次读取应先 database 再 cache。配置 save "" 与 appendonly no 禁用了持久化，但 MySQL 的笔记仍然存在。
文件：cache_restart_probe.py
import os
from note_cache import get_note, key
from redis_client import r
n = int(os.environ["NOTE_ID"])
print(get_note(n)[1])
print("cache exists", bool(r.exists(key(n))))
# 终端执行 docker restart notes-redis 后，再运行：
# print(get_note(n)[1]) -> database
# print(get_note(n)[1]) -> cache
2. 创建独立 AOF 实例
以下 Linux Bash 创建 notes-redis-aof，端口 6382，命名卷 notes_redis_aof 保存 /data。由已有配置生成 AOF 配置，主业务缓存 6380 保持原样。正式镜像应记录具体 7.x 版本与摘要，本课程用 redis:7 表示兼容主版本。
文件：终端 / 当前课时
python3 - <<'PY'
from pathlib import Path
source = Path('redis-lab/redis.conf').read_text()
source = source.replace('appendonly no', 'appendonly yes\nappendfsync everysec')
Path('redis-lab/aof.conf').write_text(source)
PY
docker run --name notes-redis-aof -d -p 127.0.0.1:6382:6379 \
-v "$PWD/redis-lab/aof.conf:/usr/local/etc/redis/redis.conf:ro" \
-v notes_redis_aof:/data \
redis:7 redis-server /usr/local/etc/redis/redis.conf
3. 验证 AOF 状态与正常重启恢复
保存为 redis_aof_probe.py。第一次运行写入样本并输出 aof_enabled、aof_last_write_status；终端 docker restart notes-redis-aof 后仅执行读取部分，不再次写入掩盖丢失。样本 TTL 10 分钟，重启后剩余时间减少；超过期限消失符合语义。
文件：redis_aof_probe.py
import os
import redis
client = redis.Redis(host="127.0.0.1",port=6382,password=os.environ["REDIS_PASSWORD"],
decode_responses=True,socket_timeout=1)
k = "notes:lab:aof-probe"
client.set(k,"persist-me",ex=600) # 只在首次运行执行，重启验证时移除此行
info = client.info("persistence")
print({name: info[name] for name in ["aof_enabled","aof_last_write_status","aof_rewrite_in_progress"]})
print("value",client.get(k),"ttl",client.ttl(k))
client.close()
4. 建立只读副本并检查复制
为 6382 的 AOF 主实例创建本机 Docker 网络，添加无持久化副本 notes-redis-replica，宿主机 6383。配置中的 masterauth 使用同一练习密码；生产应采用专用 ACL 凭据和加密通道。副本首次同步可能需要等待，INFO replication 中 master_link_status=up 才进入验收。
文件：终端 / 当前课时
docker network create notes-redis-lab
docker network connect notes-redis-lab notes-redis-aof
python3 - <<'PY'
import os
from pathlib import Path
source = Path('redis-lab/redis.conf').read_text()
source += 'replicaof notes-redis-aof 6379\n'
source += f"masterauth {os.environ['REDIS_PASSWORD']}\nreplica-read-only yes\n"
Path('redis-lab/replica.conf').write_text(source)
PY
docker run --name notes-redis-replica --network notes-redis-lab -d \
-p 127.0.0.1:6383:6379 \
-v "$PWD/redis-lab/replica.conf:/usr/local/etc/redis/redis.conf:ro" \
redis:7 redis-server /usr/local/etc/redis/redis.conf
5. 核对复制、主节点故障和恢复边界
保存为 redis_replica_probe.py。WAIT 需要在同一连接跟随写入，pipeline(transaction=False) 保持同一连接发送两个命令。返回副本确认数至少 1 后读取副本，正常条件下得到新值。随后只停止 notes-redis-aof，副本不会自动升主；检查链路转为 down 后恢复主实例。演练结束停止 aof 与 replica 两个专用实例。
文件：redis_replica_probe.py
import os
import redis
options = dict(host="127.0.0.1",password=os.environ["REDIS_PASSWORD"],decode_responses=True,
socket_timeout=2,socket_connect_timeout=1)
primary = redis.Redis(port=6382,**options)
replica = redis.Redis(port=6383,**options)
try:
print("replica",replica.info("replication")["master_link_status"])
with primary.pipeline(transaction=False) as pipe:
pipe.set("notes:lab:replica-probe","replicated",ex=600)
pipe.wait(1,1000)
written,acknowledged = pipe.execute()
print("acknowledged",acknowledged)
assert acknowledged >= 1
print("replica value",replica.get("notes:lab:replica-probe"))
finally:
primary.close()
replica.close()
# Linux 故障操作仅针对专用实例：docker stop notes-redis-aof
# 检查副本 INFO replication；不会自动升主。
# 恢复：docker start notes-redis-aof
# 演练结束：docker stop notes-redis-aof notes-redis-replica
验收结果
- 6380 重启清空缓存后，业务笔记能从 MySQL 读取并重建。
- 6382 AOF 正常重启后样本仍可读取，TTL 正常减少，持久化状态无错误。
- 副本连接正常时能读取复制值，主节点停止后不误认为已实现自动故障转移。
生产约束与故障演练
缓存节点需要维护重启，团队必须知道数据会丢在哪里、服务如何继续以及恢复验收是什么。
实现决策
- 按数据用途决定是否持久化。
- 正常重启、异常断电与主从切换分开测试。
- 复制不是数据库真源和任务幂等性的替代。
隔离环境演练
仅重启或停止 6380/6382/6383 这些本机专用实例，分别记录回源、AOF 和复制状态，再恢复或停止实验环境。
恢复验收
- 真实笔记不丢失。
- 没有宣称 everysec 零丢失。
- 未部署自动选主时明确记录为手动恢复实验。
- Redis重启只影响可重建副本；MySQL中的笔记、业务version与UTC时间保持可核对。
常见错误
启动了副本就宣称零丢失高可用
原因：忽略异步复制窗口、自动选主与客户端切换。
处理：分别验证复制延迟、故障转移组件和客户端发现；关键业务状态保存在适合其可靠性要求的真源。
重启恢复测试总成功，实际上每次都重新 SET
原因：验收脚本在读取之前又写入样本，掩盖持久化失败。
处理：将写入阶段与重启后只读验收分开，记录初始值、时间、TTL 和持久化状态。
副本长期显示 master_link_status=down
原因：Docker 网络、主实例名称或 masterauth 不匹配。
处理：检查主实例是否加入 notes-redis-lab、名称是否一致，以及双方认证配置；不要通过关掉认证规避问题。
扩展练习
为缓存实例故障写一份最短恢复顺序，并说明是否必须先恢复所有旧缓存键。
提示：先保证真源和受控回源，再逐步恢复缓存。
参考答案：
确认 MySQL 正常，启用回源限流；检查 Redis 实例身份、认证和配置；恢复连接后小批次预热热点并观察错误率与数据库延迟。可重建缓存不必恢复所有旧键。任务 Broker 需要额外检查积压、未确认任务和重复消费，不能沿用此简化流程。