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 需要额外检查积压、未确认任务和重复消费，不能沿用此简化流程。

redis-lessons.md 74 KB History Raw

Redis 缓存

01 连接 Redis、选择类型并设置 TTL

本节目标

必要概念

缓存副本与业务真源

数据类型决定命令

TTL 是数据生命期

动手实现

1. 启动只绑定本机端口的练习实例

2. 建立后续复用的客户端

3. 操作三种数据类型

4. 验证 TTL 与覆盖写入

验收结果

生产约束与故障演练

实现决策

隔离环境演练

恢复验收

常见错误

NOAUTH 或 AuthenticationError

WRONGTYPE Operation against a key holding the wrong kind of value

扩展练习

技术文档

02 实现 Cache-Aside 与带版本的 JSON 缓存

本节目标

必要概念

Cache-Aside 的读路径

两种版本解决不同问题

缓存值也需要验证

动手实现

1. 准备一条MySQL笔记

2. 显式转换MySQL结果并实现Cache-Aside

3. 对照两次读取来源

4. 注入坏格式并验证自动回源

验收结果

生产约束与故障演练

实现决策

隔离环境演练

恢复验收

常见错误

缓存返回整数pinned或没有时区的时间

修改代码增加字段后旧缓存导致 KeyError

扩展练习

技术文档

03 提交后失效缓存，并复现旧值回填竞态

本节目标

必要概念

先提交数据库，再失效缓存

旧读取可以晚于新写入回填

失效失败是待补偿工作

动手实现

1. 用受影响行数确认修改，提交后再删除缓存

2. 验证成功修改后的首次回源

3. 按确定时序重现旧值晚回填

4. 制定可检查的失效策略

验收结果

生产约束与故障演练

实现决策

隔离环境演练

恢复验收

常见错误

在异常里删缓存，用户仍看到旧值

采用延迟双删后宣称缓存强一致

扩展练习

技术文档

04 阻止缓存穿透：输入校验与短期负缓存

本节目标

必要概念

穿透是大量请求持续落到真源

空结果必须有明确标记

权限不能被共享缓存省略

动手实现

1. 实现正负结果独立编码

2. 选择确认不存在的MySQL编号并比较来源

3. 让非法编号在访问存储之前失败

4. 把创建与更新失效覆盖到负缓存

验收结果

生产约束与故障演练

实现决策

隔离环境演练

redis-lessons.md 74 KB

History Raw