agent_backend/public/downloads/redis-lessons.md

Redis 缓存

术核 AI · 后端工程课件

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

商品详情会被反复读取。先准备一个能临时保存小数据的 Redis，观察它存了什么、多久消失，再逐步接入真实商品。

从一个问题开始

商品详情会被反复读取。先准备一个能临时保存小数据的 Redis，观察它存了什么、多久消失，再逐步接入真实商品。

这次的终点

能连接本机 Redis，读写商品名称、字段和分类集合，并亲眼看到过期键消失及覆盖写入清除期限。

开始前，确认你在这里

• 先准备 Docker 与 Python 3.11+。Windows 学习者使用 WSL2，在 Linux 文件系统下准备 shop-service 目录；如果前面在 Windows 完成，复制 mysql_client.py、sql 与配置模板等练习源码，在 WSL2 重新创建 .venv，不复制 Windows 的 .venv。本课完成 Redis 连接后，第 02 课才连接 MySQL。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 第 1 步完成后，新终端依次进入 shop-service、source .venv/bin/activate、source redis-lab/local.env，再运行 python redis_client.py；这个恢复步骤复用原密码，不重新初始化 Redis。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 连接本机 6380 Redis，用 PING 验证地址与认证
• 分别用字符串、Hash 和 Set 保存商品名称、字段与去重分类
• 观察 TTL 到期消失，以及普通 SET 覆盖后期限变为 -1

一步步完成案例

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

在 Linux/WSL2 的 shop-service 根目录激活 .venv，确认 Python 3.11+、Docker 可访问，6380 与 shop-redis 属于本节专用环境。下面整块使用 Bash，不粘贴到 PowerShell。首次运行创建配置；已有本课配置时复用原密码，旧版只有 redis.conf 时从该文件恢复环境文件，不重新生成密码。Python 准备报错时先处理再继续容器命令。redis-lab 目录权限 700、local.env 权限 600，并由 .gitignore 忽略。此后每个新终端都先进入此目录、激活虚拟环境，再 source redis-lab/local.env，不用 echo 或 cat 输出密码。

文件：终端 / 当前课时

mkdir -p redis-lab
chmod 700 redis-lab
python3 - <<'PY'
import secrets
import shlex
from pathlib import Path

folder = Path('redis-lab')
config = folder / 'redis.conf'
env_file = folder / 'local.env'
if not config.exists():
    if env_file.exists():
        raise SystemExit('local.env exists without redis.conf; check this lab before continuing')
    password = secrets.token_urlsafe(32)
    config.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',
        encoding='utf-8',
    )
    config.chmod(0o644)
if not env_file.exists():
    # Also preserves the original password when resuming an older version of this lab.
    passwords = [shlex.split(line)[1] for line in config.read_text(encoding='utf-8').splitlines()
                 if line.strip().startswith('requirepass ')]
    if len(passwords) != 1:
        raise SystemExit('expected one local requirepass setting; check configuration')
    env_file.write_text(
        'export REDIS_HOST=127.0.0.1\nexport REDIS_PORT=6380\n'
        + 'export REDIS_PASSWORD=' + shlex.quote(passwords[0]) + '\n',
        encoding='utf-8',
    )
env_file.chmod(0o600)
print('local configuration ready; credentials were not printed')
PY
# If the Python preparation above reports an error, resolve it before continuing.
source redis-lab/local.env
printf 'redis-lab/\n' >> .gitignore
# The container name must already have been checked as belonging to this lesson.
if docker container inspect shop-redis >/dev/null 2>&1; then
  docker start shop-redis
else
  docker run --name shop-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
fi
python -m pip install 'redis>=5.3,<6'
python -m pip freeze > requirements-lock.txt

你应该看到什么

出现 local configuration ready，未打印凭据；首次启动返回容器 ID，恢复已有实例返回容器名。redis-lab/local.env 存在，保存原 Redis 密码；source 后第 2 步的 PING 能验证连接。

刚才发生了什么

Redis 在本课保存可丢弃的商品副本。环境变量只属于当前终端，local.env 则在本机留下下一次能读取的同一配置；source 把文件中的变量载入新 Bash。恢复学习时复用原密码，才能和现有 Redis 的认证配置一致。配置关闭数据持久化，重启后商品缓存仍需要从 MySQL 重建。

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

保存为 redis_client.py。REDIS_PASSWORD 必须存在，忘记设置时应尽早失败。客户端启动不会立即验证连通，main 中 PING 才实际访问。这里使用 redis-py 5.3 系列稳定 API，Redis 7 不支持的较新命令不在示例中使用。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_client.py，再与下方预期对照。

文件：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()

你应该看到什么

python redis_client.py 打印 ping True 和 7.x 版本；缺 REDIS_PASSWORD 时出现 KeyError，密码或连接错误时 PING 失败。

刚才发生了什么

构造客户端只是准备地址，PING 才实际发请求。超时避免连接故障让脚本一直等，decode_responses=True 让后续取回 Python 字符串。

3. 操作三种数据类型

保存为 redis_types.py，执行 python redis_types.py。所有键使用 shop:lab: 前缀，仅删除本课明确列出的三个键。反复练习不用 FLUSHALL；不同业务键不能共用同一个名字。

文件：redis_types.py

from redis_client import r

keys = ["shop:lab:types:name", "shop:lab:types:hash", "shop:lab:types:categories"]
r.delete(*keys)
r.set(keys[0], "教学水杯", ex=60)
r.hset(keys[1], mapping={"name": "教学水杯", "is_active": "false", "price_cents": "0"})
r.expire(keys[1], 60)
r.sadd(keys[2], "家居", "办公", "家居")
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)

你应该看到什么

打印商品名称、字段字典、去重后的两个分类，类型依次为 string、hash、set；结束后本节三键被删除。

刚才发生了什么

字符串适合一段值，Hash 适合按字段取值，Set 自动去重。类型决定用 GET、HGETALL 还是 SMEMBERS，同名键不能同时扮演三种类型。

4. 验证 TTL 与覆盖写入

保存为 redis_ttl.py 并运行。等待只发生在专用 1 秒练习键，不用长时间阻塞服务。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_ttl.py，再与下方预期对照。

文件：redis_ttl.py

import time
from redis_client import r

k = "shop: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)

你应该看到什么

依次看到 expires 约 1、missing -2、no expiration -1、restored expiration 约 60，最后实验键删除。

刚才发生了什么

TTL 是剩余存活秒数，-2 表示不存在，-1 表示存在但无期限。普通 SET 覆盖值也会移除原期限，所以缓存写入应同时明确新的到期策略。 SET 后 TTL 为 -1 的一幕用于认识风险；最终删除该练习键。

检查学习结果

• PING 返回 True，服务版本属于 Redis 7。
• 字符串、Hash、Set 的值和类型符合预期，Set 中 家居 只出现一次。
• 能复现 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("shop:lab:json-probe", json.dumps({"schema":1,"id":1}), ex=30)。再次 r.set 同键时仍带 ex=30，或明确使用 keepttl=True；读后用 json.loads。SET 未指定过期且未 KEEPTTL 会移除已有 TTL。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

缓存副本与业务真源

完成第 1 步“启动只绑定本机端口的练习实例”后，再回顾这条规则：本模块以MySQL 8.4 LTS的InnoDB shop_lab为商品真源，Redis只保存可重新生成的副本。缓存丢失影响读取性能，不应删除真实商品。业务连接显式RC、UTC与有界等待；数据库命令行课另行演示默认RR。计数器、锁和消息队列的正确性要求不同，不能自动套用可随时丢弃的缓存策略。

数据类型决定命令

完成第 3 步“操作三种数据类型”后，再回顾这条规则：字符串可保存序列化 JSON；Hash 保存多个字段；Set 表示无重复成员。不同类型有各自命令，GET 一个 Hash 会返回 WRONGTYPE。这里不使用额外 RedisJSON 模块，JSON 文本由 Python 编解码。

TTL 是数据生命期

完成第 4 步“验证 TTL 与覆盖写入”后，再回顾这条规则：TTL 正数表示剩余秒数，-1 表示存在但没有过期时间，-2 表示键不存在。SET 默认覆盖值并移除原 TTL，除非显式设置新期限或保留期限。缓存写入通常把值和 EX/PX 放在同一条 SET 中，避免写值成功却未设置过期。

技术文档

• Redis：数据类型
• Redis：SET
• Redis：TTL
• redis-py 5.3：连接参数

---

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

同一件商品连读两次时，希望第二次不用再查 MySQL。先观察第一次从数据库取、第二次从 Redis 取，再处理坏缓存。

从一个问题开始

同一件商品连读两次时，希望第二次不用再查 MySQL。先观察第一次从数据库取、第二次从 Redis 取，再处理坏缓存。

这次的终点

同一商品第一次来源 database、第二次 cache；缓存格式错误时自动回源修复，商品价格和 UTC 时间保持一致。

开始前，确认你在这里

• 已完成 Redis 第 01 课和数据库第 03 课；shop-service 中有 redis_client.py、mysql_client.py，在同一终端确认 Redis PING 成功且 MySQL 连接 shop_lab，保留两端本机密码环境变量。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 新终端先 source redis-lab/local.env。第 02 课会把 MYSQL_* 和 PRODUCT_ID 一起保存在同一受限文件，后续无需重新输入密码或猜商品编号；不要重新运行初始化来生成新 Redis 密码。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 为真实 MySQL 商品准备可复用的编号与 JSON 缓存编码
• 对比首次 database 与第二次 cache 的读取来源和内容
• 注入坏 JSON 与不支持的结构版本，验证回源后缓存得到修复

一步步完成案例

1. 准备一条MySQL商品

在当前 Linux/WSL2 的 shop-service 根目录保留数据库第 03 课 mysql_client.py，安装其 PyMySQL 依赖，并 source redis-lab/local.env。将下面完整代码保存为 seed_cache_product.py，运行 python seed_cache_product.py；首次未设置 MYSQL_PASSWORD 时，在隐藏输入提示中输入已有 app_user 的本机密码。脚本先确认连接库为 shoplab，再插入专用商品，并把 MYSQL* 和实际 PRODUCT_ID 合并保存到受限 local.env，保留 Redis 原密码。成功后再次 source redis-lab/local.env，再进入下一步；连接失败先回数据库连接课排查。

文件：seed_cache_product.py

import getpass
import os
import shlex
from pathlib import Path
from mysql_client import connect

env_file = Path('redis-lab/local.env')
if not env_file.is_file():
    raise SystemExit('complete Redis lesson 01 and source redis-lab/local.env first')
defaults = {'MYSQL_HOST': '127.0.0.1', 'MYSQL_PORT': '3307',
            'MYSQL_DATABASE': 'shop_lab', 'MYSQL_USER': 'app_user'}
for name, value in defaults.items():
    os.environ.setdefault(name, value)
if not os.environ.get('MYSQL_PASSWORD'):
    os.environ['MYSQL_PASSWORD'] = getpass.getpass('Existing local MySQL app_user password: ')

conn = connect()
try:
    with conn.cursor() as cur:
        cur.execute('SELECT DATABASE() AS database_name')
        if cur.fetchone()['database_name'] != 'shop_lab':
            raise RuntimeError('this sample requires the local shop_lab database')
        cur.execute('INSERT INTO products(name,description,price_cents) VALUES(%s,%s,%s)',
                    ('教学水杯', '数据库保存商品真实信息', 1990))
        product_id = cur.lastrowid
    conn.commit()
    print('PRODUCT_ID=', product_id)
except BaseException:
    conn.rollback()
    raise
finally:
    conn.close()

# Preserve Redis credentials, and replace only this lesson's MySQL settings and sample ID.
names = [*defaults, 'MYSQL_PASSWORD', 'PRODUCT_ID']
saved = [line for line in env_file.read_text(encoding='utf-8').splitlines()
         if not any(line.startswith(f'export {name}=') for name in names)]
os.environ['PRODUCT_ID'] = str(product_id)
saved.extend(f'export {name}={shlex.quote(os.environ[name])}' for name in names)
env_file.write_text('\n'.join(saved) + '\n', encoding='utf-8')
env_file.chmod(0o600)
print('local settings saved; run: source redis-lab/local.env')

你应该看到什么

脚本输出真实 PRODUCT_ID 和 local settings saved，不回显 MySQL 密码；source redis-lab/local.env 后，新脚本可读取同一编号、Redis 与 MySQL 配置，数据库存在价格 1990 的教学水杯。

刚才发生了什么

数据库返回的主键把后续实验固定到一条真实商品。getpass 让第一次输入密码时不显示在屏幕上；保存时按 Bash 的引号规则编码，所以密码里的特殊字符不会变成 Shell 命令。变量的文件留存与当前终端加载是两件事，写完文件后再 source，当前 Shell 才能得到新的 PRODUCT_ID。

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

保存product_cache.py，完整替换旧实现。PyMySQL读取DATETIME(6)通常得到无tzinfo的datetime，本项目确认所有写入都采用UTC后才附加timezone.utc；BOOLEAN的0/1显式转bool。此教学回源每次新建连接，生产改用数据库第10课的有界池或API模块对应异步连接池。商品价格统一使用0..2147483647 范围的整数 price_cents（分）；缓存编码、解码和数据库查询均包含此字段，不能把布尔值或浮点金额当作合法价格。

文件：product_cache.py

import json
from datetime import datetime,timezone
from mysql_client import connect
from redis_client import r

def key(product_id):
    return f"shop:lab:v1:product:{{{product_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_product(row):
    if row is None:
        return None
    return {
        "id":int(row["id"]),"name":row["name"],"description":row["description"],
        "is_active":bool(row["is_active"]),
        "price_cents":int(row["price_cents"]),
        "created_at":utc_iso(row["created_at"]),
        "updated_at":utc_iso(row["updated_at"]),
        "version":int(row["version"]),
    }

def load_product(product_id):
    conn=connect()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT id,name,description,is_active,price_cents,created_at,updated_at,version FROM products WHERE id=%s",(product_id,))
            row=cur.fetchone()
        conn.commit()
        return normalize_product(row)
    except BaseException:
        conn.rollback()
        raise
    finally:
        conn.close()

def encode(product):
    return json.dumps({"schema":1,"product":product},ensure_ascii=False)

def decode(raw,product_id):
    data=json.loads(raw)
    if not isinstance(data,dict) or data.get("schema")!=1:
        raise ValueError("unsupported cache schema")
    product=data["product"]
    if not isinstance(product,dict) or type(product.get("id")) is not int or product["id"]!=product_id:
        raise ValueError("invalid cached product id")
    if not isinstance(product.get("name"),str) or not 1<=len(product["name"])<=120:
        raise ValueError("invalid cached name")
    if not isinstance(product.get("description"),str) or len(product["description"])>16000:
        raise ValueError("invalid cached description")
    if type(product.get("is_active")) is not bool or type(product.get("version")) is not int or product["version"]<1:
        raise ValueError("invalid cached fields")
    if type(product.get("price_cents")) is not int or not 0 <= product["price_cents"] <= 2147483647:
        raise ValueError("invalid cached price_cents")
    for field in ("created_at","updated_at"):
        parsed=datetime.fromisoformat(product[field])
        if parsed.tzinfo is None or parsed.utcoffset().total_seconds()!=0:
            raise ValueError("cached timestamp must be UTC")
    return product

def get_product(product_id):
    raw=r.get(key(product_id))
    if raw is not None:
        try:
            return decode(raw,product_id),"cache"
        except (ValueError,KeyError,TypeError):
            r.delete(key(product_id))
    product=load_product(product_id)
    if product is not None:
        r.set(key(product_id),encode(product),ex=60)
    return product,"database"

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。product_cache.py 可导入，包含 load_product、encode/decode 与 get_product；仅定义函数不会自动写缓存，下一步才调用读路径。

刚才发生了什么

读路径先找缓存，没找到才查询 MySQL，并把查询结果存 60 秒，这叫 Cache-Aside。日期和布尔值先转成明确 JSON 形式，避免驱动返回的 Python 类型让缓存编码失败。 商品描述上限16000字符，与API保持一致。读取成功也commit结束RC事务，异常rollback，finally关闭。

3. 对照两次读取来源

保存为 cache_read.py。只清除当前练习商品的一个缓存键。没有记录时返回 None，本课尚未缓存不存在的结果。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_read.py，再与下方预期对照。

文件：cache_read.py

import os
from redis_client import r
from product_cache import key, get_product

product_id = int(os.environ["PRODUCT_ID"])
r.delete(key(product_id))
first, source1 = get_product(product_id)
second, source2 = get_product(product_id)
assert first == second and first is not None
print(source1, source2, r.ttl(key(product_id)))
print(first)

你应该看到什么

python cache_read.py 打印 database cache 和 1–60 范围的 TTL，两份商品内容相等且包含当前真实 id。

刚才发生了什么

第一次删除了指定键，必然回源；第二次命中刚写入的副本。来源标签让“少查一次数据库”成为能直接观察的行为。 第一次来源 database，第二次 cache，内容应相等，TTL 在 1 到 60 之间。

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

先写入不可解析文本，再写入不受支持的 schema。每次调用都应返回 database 和合法商品，然后缓存被合法 JSON 替换。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_bad_format.py，再与下方预期对照。

文件：cache_bad_format.py

import os
from redis_client import r
from product_cache import key, get_product

product_id = int(os.environ["PRODUCT_ID"])
for bad in ["not-json", '{"schema":99,"product":{"id":1}}']:
    r.set(key(product_id), bad, ex=60)
    product, source = get_product(product_id)
    print(source, product["id"])
    assert source == "database"

你应该看到什么

两种坏值都打印 database 和当前商品编号，缓存被 schema=1 的合法 JSON 替换；无 JSON 解码异常泄漏到使用方。

刚才发生了什么

缓存可能过时或被错误写入，所以命中后仍要验证格式。schema 表示缓存结构版本，商品 version 表示业务数据版本，两者解决不同问题。 Redis 网络错误尚未在此函数内处理，第 06 和 09 课补齐有界降级。

检查学习结果

• 第一次返回 database，第二次返回 cache，业务字段一致。
• 缓存含schema和整数product.version；is_active是真正JSON布尔值，时间为+00:00的ISO8601。
• 坏 JSON 和未知 schema 不直接导致业务解码崩溃，回源后得到有效内容。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

同一商品被反复读取，希望降低数据库访问同时保留格式升级能力。

实现决策

• MySQL 成功读取后才回填。
• 缓存格式版本和业务版本分离。
• 异常与不存在使用不同结果。

隔离环境演练

给单个练习键写入坏 JSON 和未知 schema，检查回源和重新编码。

恢复验收

• 命中来源可观测。
• 无任意对象反序列化。
• 缓存值不包含未设计对外展示的敏感字段。

遇到问题时查这里

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

原因：直接序列化驱动结果，没有把MySQL类型转成API约定的DTO。

处理：在normalize_product中集中bool、整数与UTC时间转换；对DATETIME补UTC依赖明确存值约定，不能猜测已有未知来源时间。

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

原因：缓存格式没有版本和校验，部署后仍读取旧结构。

处理：提升缓存 schema 或键名前缀，允许旧键自然过期；读取方明确接受的结构。

自己试一次

让记录不存在时也能返回明确来源，并说明为什么不能把数据库异常当成不存在。

提示：None 表示成功查询且无行；连接异常代表没有完成查询。

参考答案：

load_product成功查询且fetchone为None时，返回None与database来源。PyMySQL连接异常、语句超时和数据转换异常继续传播，不能伪造不存在并写负缓存。读取事务也显式commit或rollback并关闭；故障降级不意味着吞掉真源错误。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

Cache-Aside 的读路径

完成第 2 步“显式转换MySQL结果并实现Cache-Aside”后，再回顾这条规则：应用先 GET；不存在时查询数据库，再把结果 SET 到缓存。Redis 不会自动知道 MySQL 更新。先把基本路径做对，后续逐课处理失效、并发回填和故障降级。

两种版本解决不同问题

完成第 4 步“注入坏格式并验证自动回源”后，再回顾这条规则：缓存外层schema=1表示JSON协议，product.version是MySQL记录的业务版本。MySQL不会自动把DATETIME附带UTC信息，也不会把TINYINT(1)自动转成JSON布尔值，所以序列化边界显式转换。将来添加JSON列时还应确认驱动返回的是文本还是已解码对象，不能让多个入口各自猜测。

缓存值也需要验证

完成第 4 步“注入坏格式并验证自动回源”后，再回顾这条规则：反序列化 JSON 后还要检查结构和编号，坏缓存应被当作未命中处理。数据库时间转换为带时区 ISO 8601，不用 pickle 反序列化来自缓存的任意对象。本课缓存只包含列表出的商品字段。

技术文档

• Redis：GET
• Redis：Python 客户端
• Python：json
• MySQL8.4：日期时间类型
• PyMySQL：DictCursor与显式提交

---

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

把商品名称改了，页面为什么仍可能读到旧名？先实现提交后清缓存，再手工安排两次读取的先后顺序复现旧值回来。

从一个问题开始

把商品名称改了，页面为什么仍可能读到旧名？先实现提交后清缓存，再手工安排两次读取的先后顺序复现旧值回来。

这次的终点

成功修改后首次读取回源得到新版本；还能稳定复现并解释旧读取晚回填，写出可核对的失效策略。

开始前，确认你在这里

• 已完成 Redis 第 02 课，当前终端有 PRODUCT_ID；在 shop-service 确认 load_product 能查到商品，Redis 与 MySQL 均可用，保留 product_cache.py。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 新终端先 source redis-lab/local.env。第 02 课会把 MYSQL_* 和 PRODUCT_ID 一起保存在同一受限文件，后续无需重新输入密码或猜商品编号；不要重新运行初始化来生成新 Redis 密码。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 用当前版本更新商品，在数据库提交成功后失效缓存
• 验证修改后的第一次读取回源并取得新版本
• 按固定时序复现旧值晚回填，并列出失效失败的补偿与观测项

一步步完成案例

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

保存cache_write.py。MySQL没有此处的DML RETURNING，先用expected_version条件UPDATE并检查rowcount，再在同一持锁事务SELECT真实结果。显式conn.commit成功后关闭连接，再调用Redis DEL；commit网络错误时结果可能不确定，不能盲重试。示例invalidated=False需由生产Outbox补偿，此文件未实现可靠投递。

文件：cache_write.py

from mysql_client import connect
from redis.exceptions import RedisError
from product_cache import key
from redis_client import r

def rename_product(product_id,name,expected_version):
    conn=connect()
    try:
        with conn.cursor() as cur:
            cur.execute("UPDATE products SET name=%s,version=version+1,updated_at=UTC_TIMESTAMP(6) WHERE id=%s AND version=%s",
                        (name,product_id,expected_version))
            if cur.rowcount!=1:
                raise ValueError("not found or version conflict")
            cur.execute("SELECT id,name,price_cents,version FROM products WHERE id=%s",(product_id,))
            row=cur.fetchone()
        conn.commit()
    except BaseException:
        conn.rollback()
        raise
    finally:
        conn.close()
    try:
        r.delete(key(product_id))
        invalidated=True
    except RedisError:
        invalidated=False
    return {"id":int(row["id"]),"name":row["name"],"price_cents":int(row["price_cents"]),"version":int(row["version"]),"invalidated":invalidated}

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。rename_product 返回新名称、递增 version 和 invalidated 标志；版本不匹配抛 ValueError，不会删除缓存假装成功。

刚才发生了什么

条件 UPDATE 先证明改的是读到的版本，commit 后才删除缓存。若先删再提交，别的请求可能在新值还不可见时把旧值又放回去。 UPDATE总是递增version，成功匹配必改变行。

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

保存为 cache_invalidate.py 并运行。先预热旧值，再以当前版本更新。更新后键不存在，下一次 get_product 返回数据库的新名称与新版本。重复运行时从数据库读取当前版本，不假定总是 1。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_invalidate.py，再与下方预期对照。

文件：cache_invalidate.py

import os
from product_cache import get_product, key, load_product
from cache_write import rename_product
from redis_client import r

n = int(os.environ["PRODUCT_ID"])
before = load_product(n)
r.delete(key(n))
get_product(n)
print(rename_product(n, "cache03-new-name", before["version"]))
assert r.get(key(n)) is None
after, source = get_product(n)
print(source, after["name"], after["version"])
assert source == "database" and after["version"] == before["version"] + 1

你应该看到什么

打印 invalidated=True，修改后键不存在；下一次打印 database、新名称和比原来大 1 的版本。

刚才发生了什么

失效是让旧副本失去效力，不是把数据库再写一次。下一位读者会负责重新填充已经提交的新值。

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

本脚本将两个并发请求的步骤按顺序手工调度，结果不依赖线程调度或 sleep。只操作 PRODUCT_ID 专用商品。A 先读取旧值并暂停，B 完成提交和失效，A 最后回填旧值；读取命中旧版本，数据库版本更高。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_stale_refill.py，再与下方预期对照。

文件：cache_stale_refill.py

import os
from product_cache import load_product, get_product, encode, key
from cache_write import rename_product
from redis_client import r

n = int(os.environ["PRODUCT_ID"])
r.delete(key(n))
snapshot_a = load_product(n)  # A: 已回源，尚未回填
written_b = rename_product(n, "cache03-race-new", snapshot_a["version"])  # B
r.set(key(n), encode(snapshot_a), ex=5)  # A: 旧值晚到
cached, source = get_product(n)
fresh = load_product(n)
print("cached", source, cached["version"], "database", fresh["version"])
assert cached["version"] < fresh["version"]
r.delete(key(n))  # 清理本次故障注入
assert get_product(n)[0]["version"] == fresh["version"]

你应该看到什么

输出 cached cache 的版本小于 database 版本，断言通过；删掉故障键再读取后两者版本一致。

刚才发生了什么

A 已经拿到旧快照，B 提交并删缓存后，A 仍可晚到回填；这就是交错时序造成的竞态。提交后删除降低常见陈旧读，但不能单凭它承诺强一致。

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

为 shop-service 的详情、列表、分类聚合分别列出受一次编辑影响的键。详情可精确删除；列表缓存需要版本化命名空间或短 TTL，不能用 KEYS 模糊删除整个实例。将下列记录保存为开发验收清单，后续 Outbox 模块实现重试。

文件：终端 / 当前课时

事件：product.updated，字段 product_id / committed_version / event_id
写入：业务 UPDATE 与 outbox INSERT 在同一个MySQL InnoDB事务（事件表DDL在部署阶段预先完成）
消费：按事件删除详情键及明确的派生键，成功后记录处理状态
重复：DEL 对不存在键仍可安全重放
失败：有限退避后进入待处理队列并告警，不丢弃失效事件
读取：写接口返回同事务SELECT得到并已commit的真实版本；强一致读取显式绕过缓存
观测：失效失败数、最老未处理事件年龄、回源率、旧版本样本

你应该看到什么

失效清单列出详情和派生键、事件字段、失败处理及强一致读取路径；明确哪些补偿尚待 Outbox 实现。

刚才发生了什么

删除失败是一项未完成工作，不能因商品写成功就丢掉。把失效事件与商品更新共同落库，再有界重试，可让短暂 Redis 故障之后仍有恢复依据。

检查学习结果

• DEL 只出现在数据库事务成功退出后。
• 固定时序实验能得到缓存版本小于数据库版本，清理后恢复。
• 设计文档区分提交成功与失效成功，并给失效失败安排可靠补偿。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

编辑提交成功后用户立即刷新，后台又可能存在较慢的并发读取。

实现决策

• 提交后删除详情缓存。
• 用确定性时序测试陈旧回填。
• 可靠失效事件与直接读写结果满足不同一致性需求。
• 事务中完成UPDATE与结果SELECT，commit成功后再失效；迁移DDL不混入业务事务。

隔离环境演练

仅对 PRODUCT_ID 练习商品模拟旧读取在新提交之后回填，比较两个版本并清理故障值。

恢复验收

• 没有宣称 Cache-Aside 天然强一致。
• 失效故障不会伪装成数据库回滚。
• 派生缓存清单和重试责任明确。

遇到问题时查这里

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

原因：失效顺序没有跟随真实提交结果，或遗漏派生列表键。

处理：围绕事务提交定义事件，列出所有派生缓存；不要以请求函数是否执行完代替提交状态。

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

原因：延迟长短无法覆盖任意暂停、网络延迟和进程故障。

处理：把延迟删除视为减少窗口的措施；依据业务选择可接受陈旧、可靠事件或直接读真源。

自己试一次

数据库更新已提交但 Redis 断开时，接口能否返回“数据库操作已回滚”？应记录什么？

提示：数据库提交和 Redis 失效是两个独立结果。

参考答案：

不能报告回滚。commit已成功时应保留实际提交版本，记录待补偿事件及失败原因，使用同事务Outbox可靠重试缓存删除。若commit本身因网络中断返回错误，结果是不确定，需按幂等标识查询真源核对，不能假定没提交。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

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

完成第 1 步“用受影响行数确认修改，提交后再删除缓存”后，再回顾这条规则：数据库提交成功后删除详情缓存，下一次读取从真源重建。如果在提交前删除，另一个读请求可能读到尚未更新的数据库旧值并重新缓存。提交后删除减少这一类窗口，但不消除所有并发竞态。

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

完成第 3 步“按确定时序重现旧值晚回填”后，再回顾这条规则：读者 A 在写者提交前读到 v1，写者 B 提交 v2 并删除缓存，A 随后把 v1 回填。即使删除发生在提交后，缓存仍可能短暂陈旧。TTL 只限制一次缓存副本的存活时间，不应被表述为完整一致性证明。

失效失败是待补偿工作

完成第 4 步“制定可检查的失效策略”后，再回顾这条规则：MySQL事务和Redis命令没有共同本地事务。数据库已提交、DEL失败不能报告回滚。生产可在同一InnoDB事务写业务行和Outbox（事件表提前由独立DDL创建），后台可靠重试删除；若要求立即读取自己提交的结果，直接返回同一事务SELECT得到的版本或显式绕过缓存。

技术文档

• Redis：DEL
• Redis：客户端缓存与失效
• MySQL8.4：InnoDB事务
• MySQL8.4：UPDATE

---

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

有人不断查询不存在的商品。普通缓存只存查到的商品，每次落空仍会查 MySQL；本课短暂记住“确实没有”。

从一个问题开始

有人不断查询不存在的商品。普通缓存只存查到的商品，每次落空仍会查 MySQL；本课短暂记住“确实没有”。

这次的终点

不存在编号第一次查数据库、第二次命中 5 秒负缓存；非法编号在访问存储前失败，新建或更新可失效相关键。

开始前，确认你在这里

• 已完成 Redis 第 02–03 课，在 shop-service 保留 product_cache.py、redis_client.py、mysql_client.py；确认当前连接都是本机练习实例。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 新终端先 source redis-lab/local.env。第 02 课会把 MYSQL_* 和 PRODUCT_ID 一起保存在同一受限文件，后续无需重新输入密码或猜商品编号；不要重新运行初始化来生成新 Redis 密码。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 用独立编码短暂缓存数据库确认不存在的商品
• 验证不存在编号从 database 转为 negative-cache，期限为五秒
• 在访问存储前拒绝非法编号，并把写入失效扩展到负缓存

一步步完成案例

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

保存为 negative_cache.py。单独使用 v1:lookup 命名空间，不与第 02 课 JSON 结构混用。类型校验拒绝 bool 作为整数编号；上限 2^63-1 对应数据库 bigint 的正数范围。

文件：negative_cache.py

import json
from redis_client import r
from product_cache import load_product,decode,encode

def lookup_key(product_id):
    return f"shop:lab:v1:lookup:{{{product_id}}}"

def get_with_negative(product_id):
    if type(product_id) is not int or not 1 <= product_id <= 2**63 - 1:
        raise ValueError("invalid product id")
    k = lookup_key(product_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"
            product = data["product"]
            if not isinstance(product, dict) or product.get("id") != product_id:
                raise ValueError("bad product")
            return decode(encode(product),product_id), "cache"
        except (ValueError, KeyError, TypeError):
            r.delete(k)
    product = load_product(product_id)  # 异常继续向外传递，不伪造不存在
    found = product is not None
    payload = {"schema": 1, "found": found, "product": product}
    r.set(k, json.dumps(payload), ex=60 if found else 5)
    return product, "database"

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。negative_cache.py 定义独立 lookup 键；正常商品缓存 60 秒，无记录用 found=False 保存 5 秒，读取异常不会编码为不存在。

刚才发生了什么

负缓存不是负数，而是把“查过且没有”也临时存下来。显式 found 标记把无记录与坏格式、网络错误分开，短期限减少后来新建商品被旧缺失结论挡住的时间。 正结果 60 秒，确定无行 5 秒。

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

只在独立练习库选MAX(id)+100000并确认在BIGINT正数范围。MAX这里只是产生确定不存在的实验输入，不用于获取刚插入记录的主键。成功只读事务也显式commit，异常rollback，最终close。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python negative_probe.py，再与下方预期对照。

文件：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 products")
        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))

你应该看到什么

第一次打印 (None, 'database')，第二次为 (None, 'negative-cache')，TTL 约 1–5 秒，最后删除该实验键。

刚才发生了什么

这两次结果内容相同，但第二次不用查 MySQL。只有成功查询确定没有行才能得到这个结论，数据库连接失败不能伪装为商品不存在。

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

保存并运行。bad 值都会触发 ValueError；后续 API 接入时由参数模型先执行同等校验。单个非法值的验证不要发起 Redis 或 MySQL 请求。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python negative_invalid.py，再与下方预期对照。

文件：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")

你应该看到什么

五种 bad 输入均打印 invalid product id，脚本不触发 else 断言；这些调用在 Redis GET 之前已退出。

刚才发生了什么

先做类型和范围检查，能挡住连数据库都不可能存在的编号。Python 的 bool 是 int 子类，因此用 type(...) is int 明确拒绝 True。

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

将 lookup_key 对应键加入提交后失效清单。新建时数据库返回 id 后再删除同一编号的负缓存；更新与删除时同时失效详情和 lookup 两种命名空间。下面函数只演示失效动作，调用必须放在业务事务提交后。

文件：cache_invalidate_keys.py

from redis_client import r
from product_cache import key
from negative_cache import lookup_key

def invalidate_product_caches(product_id):
    return r.delete(key(product_id), lookup_key(product_id))

# 调用位置：数据库提交成功后，或可靠 Outbox 消费函数中。
# 不在这里捕获错误后无声丢弃。

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。提交后调用失效函数可同时删除详情键和 lookup 键，返回删除数量；本段只定义动作，尚未自动挂入所有写路径。

刚才发生了什么

如果新建商品之前有人缓存过“没有”，只删除正缓存会留下错误结论。创建、更新与删除要一起维护所有派生读路径，调用仍放在数据库提交成功之后。

检查学习结果

• 无行结果只回源一次，短期第二次返回 negative-cache。
• 非法编号在存储访问之前被拒绝。
• 新建、修改和删除的失效清单包含负缓存键，数据库错误不写入负缓存。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

重复访问不存在商品的请求增加，数据库空查询消耗开始明显。

实现决策

• 格式校验先于缓存读取。
• 只缓存确定的无行结果，负 TTL 短于正常缓存。
• 负缓存结合入口限流与容量监控。

隔离环境演练

对确定不存在的练习编号连续读取，再提交非法编号，比较来源和存储请求。

恢复验收

• 不存在与故障分开。
• 没有以负缓存绕过鉴权。
• 新增路径失效负缓存。

遇到问题时查这里

新建商品后短时间仍显示不存在

原因：创建前留下负缓存，创建后没有失效。

处理：用数据库返回的真实 id 删除对应负缓存，并设置较短负 TTL。

缓存里出现海量随机不存在编号

原因：攻击者每次换编号，短 TTL 也无法阻止瞬时键膨胀。

处理：在入口鉴权和限流，限制参数与访问范围；观察负缓存创建率和内存，不依赖负缓存单独抗攻击。

自己试一次

如果增加布隆过滤器，返回“可能存在”和“确定不在过滤器”各意味着什么？

提示：过滤器有误判，且构建及更新是否完整决定能否用来拒绝查询。

参考答案：

可能存在仍需查缓存或数据库。确定不在过滤器只有在过滤器完整且新增写入同步维护时才可拒绝；过滤器延迟会误拒绝新记录，因此需要回源或版本化切换策略。它不能代替权限检查与数据库真源。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

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

完成第 2 步“选择确认不存在的MySQL编号并比较来源”后，再回顾这条规则：查询从不存在的编号时，普通 Cache-Aside 每次都得到空结果而不回填，会重复访问数据库。输入范围校验先拦掉无意义请求，负缓存减少同一个合法但不存在编号的重复查询。

空结果必须有明确标记

完成第 1 步“实现正负结果独立编码”后，再回顾这条规则：Redis GET 返回 None 表示缓存键不存在，不等于业务记录不存在。负缓存使用 found=false；正缓存使用 found=true 和 product，读取时分别处理。数据库连接失败绝不能被编码为 found=false。

权限不能被共享缓存省略

完成第 4 步“把创建与更新失效覆盖到负缓存”后，再回顾这条规则：记录缓存只是数据副本，不代替鉴权。多租户应用应先完成主体与租户校验，并把需要隔离的租户放进键名。单用户课程暂不引入租户字段；不能把某个用户的 403 缓存成所有用户都不存在。

技术文档

• Redis：过期键
• Redis：布隆过滤器
• Redis：SCAN

---

05 抑制热点击穿：带令牌的单飞回填

热门商品的缓存刚过期，八个读取同时到来。如果它们都查数据库，就重复做了同一件事；让其中一个负责填充，其余短等结果。

从一个问题开始

热门商品的缓存刚过期，八个读取同时到来。如果它们都查数据库，就重复做了同一件事；让其中一个负责填充，其余短等结果。

这次的终点

八个并发读取在本机正常负载下只回源一次；旧填充者不能删除新持有者的锁，超过等待预算明确失败。

开始前，确认你在这里

• 已完成 Redis 第 02–04 课；在 shop-service 终端确认 PRODUCT_ID 对应商品存在，Redis 与 MySQL 正常，保留 product_cache.py。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 新终端先 source redis-lab/local.env。第 02 课会把 MYSQL_* 和 PRODUCT_ID 一起保存在同一受限文件，后续无需重新输入密码或猜商品编号；不要重新运行初始化来生成新 Redis 密码。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 为同一商品的缓存填充设置短租约、令牌和有限等待
• 用八个并发读取统计实际回源次数，正常情况下只查询一次
• 验证旧令牌不能释放新锁，过期填充者不能发布缓存

一步步完成案例

1. 保存安全释放脚本

保存为 cache_singleflight.py，后续步骤继续向此文件追加。键名使用同一个 {product_id} 哈希标签，便于将来多键脚本要求同槽；当前未部署 Redis Cluster。脚本只有固定数量短操作，不在 Redis 内做长循环或外部访问。

文件：cache_singleflight.py

import secrets
import time
from redis_client import r
from product_cache import key, load_product, 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
""")

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。cache_singleflight.py 中注册 release/publish 两段脚本；只有 token 相等时才删锁或发布缓存，此时尚未发起回源。

刚才发生了什么

token 是本次填充者的随机凭证。Lua 脚本把“比较凭证”和“修改键”连续执行，避免检查之后所有者变化却仍然删除别人的锁。

2. 实现有限等待与二次读取

追加到 cache_singleflight.py。不能让所有竞争者超时后同时无限回源。

文件：cache_singleflight.py

def singleflight_product(product_id, loader=load_product):
    cache_key, lock_key = key(product_id), key(product_id) + ":fill"
    raw = r.get(cache_key)
    if raw is not None:
        return decode(raw, product_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, product_id)
            product = loader(product_id)
            if product is None:
                return None
            stored = publish(keys=[lock_key, cache_key], args=[token, encode(product), 60000])
            if not stored:
                raise CacheBusy("fill lease expired")
            return product
        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, product_id)
    raise CacheBusy("cache fill still running")

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。抢到填充权时再次检查缓存；没抢到的最多等 0.3 秒，仍无结果则 CacheBusy；填充租约过期时也不发布。

刚才发生了什么

SET NX 只在键不存在时写入，所以一个时刻只有一个当前持有者。二次检查避免等待期间别人已填好又重复查询；等待上限让热门请求不会无限堆住。 获得锁后再读一次缓存，避免等待途中别人已经回填。竞争失败最多等 0.3 秒，每 20 毫秒看一次结果；耗尽后 CacheBusy，应映射为受控 503 或按策略返回允许的旧值。

3. 并发验证回源次数

保存为 cache_singleflight_probe.py。8 个线程同时读取同一商品，模拟回源增加 80 毫秒。正常本机负载下 calls 应为 1；若发生 Redis 超时或调度超过等待期限，应把失败视为实际故障记录，不删除断言掩盖问题。专用记录必须存在，运行前清理详情和填充锁。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_singleflight_probe.py，再与下方预期对照。

文件：cache_singleflight_probe.py

import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from product_cache import key, load_product
from redis_client import r
from cache_singleflight import singleflight_product

n = int(os.environ["PRODUCT_ID"])
r.delete(key(n), key(n) + ":fill")
guard = threading.Lock()
calls = 0

def counted_loader(product_id):
    global calls
    with guard:
        calls += 1
    time.sleep(0.08)
    return load_product(product_id)

with ThreadPoolExecutor(max_workers=8) as workers:
    rows = list(workers.map(lambda _: singleflight_product(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)

你应该看到什么

python cache_singleflight_probe.py 打印 database loads 1，八个结果均为同一商品；超时则保留实际错误供排查。

刚才发生了什么

计数包在真实 loader 外面，记录实际回源次数。八个读者复用同一填充结果，这种合并同一时刻重复工作的方式叫单飞。

4. 验证旧令牌不能释放新锁

保存为 cache_lock_token.py。只对本课 token-probe 键模拟替换所有者，原令牌释放返回 0，键仍为 new-owner。这个实验直接设置新 token 以固定复现所有权变化；真实场景通常由原锁过期后竞争产生。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_lock_token.py，再与下方预期对照。

文件：cache_lock_token.py

from redis_client import r
from cache_singleflight import release

k = "shop: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")

你应该看到什么

打印 ownership check passed；旧 token 释放返回 0，新 token 对应锁仍存在，当前持有者释放返回 1。

刚才发生了什么

锁会过期，旧工作也可能还在运行，因此“我曾拿到锁”不足以证明“现在仍是我”。释放和发布都检查当前令牌，防止晚到动作破坏新持有者。

检查学习结果

• 正常热点并发实验只回源一次，其余读取缓存结果。
• 旧 token 不能删除新 token 的锁。
• 等待、租期、Redis 超时各有上限，锁过期后不发布旧持有者结果。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

热门商品缓存到期后突然出现大量相同详情请求。

实现决策

• 先用进程内合并或单实例填充锁降低重复回源。
• 令牌检查与发布、释放原子执行。
• 竞争失败有界等待，不把所有失败者直接推向数据库。

隔离环境演练

在本机令牌键替换所有者，再延长 loader 到超过租期，检查旧持有者行为。

恢复验收

• 旧持有者不能删新锁。
• 数据库回源次数有度量。
• 不把缓存锁当作生产业务唯一性保证。

遇到问题时查这里

finally 中 DEL 删除了别人的锁

原因：原锁过期后另一请求获得锁，旧请求无条件删除。

处理：使用 Lua 比较 token 并原子删除，不能把 GET 与 DEL 分开发。

把缓存锁用于导出结果发布并宣称绝不重复

原因：租期过期、进程暂停或异步复制故障会破坏单持有者假设。

处理：业务正确性使用数据库约束、幂等键或具备 fencing 校验的资源；缓存锁只减少重复回源。

自己试一次

把 loader 延迟改为 2.2 秒，预测原持有者和竞争者结果，并说明生产应如何处理。

提示：填充租期 2 秒，竞争者等待 0.3 秒；二者是不同预算。

参考答案：

竞争者先收到 CacheBusy；原持有者回源完成后 publish 因 token 已过期返回 0，也抛出 CacheBusy。应该限制回源耗时、选择合适租期并监控失败；若需要续租必须验证所有者且限定总时间，不能无限延长。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

单飞用于压低重复工作

完成第 3 步“并发验证回源次数”后，再回顾这条规则：同一个热点缓存失效时，许多请求同时回源会压垮数据库。短期填充锁让一个请求重建，其他请求短暂等待缓存或收到忙响应。它服务于缓存效率，不承诺跨故障条件下业务副作用严格只执行一次。

释放时检查所有者令牌

完成第 1 步“保存安全释放脚本”后，再回顾这条规则：每次竞争生成随机 token，并用 SET lock token NX PX 同时设置所有权和租期。释放必须在一段 Lua 中比较 token 再删除；先 GET 后 DEL 存在两条命令之间被新持有者接手的窗口。Redis 7 示例使用 Lua，不使用较新版本的条件删除命令。

过期意味着可能出现另一位持有者

完成第 4 步“验证旧令牌不能释放新锁”后，再回顾这条规则：进程暂停或回源过慢会让锁先过期。旧持有者恢复后不能删除或发布覆盖新持有者结果。下面脚本把校验、回填与释放放在一次 Lua 中；数据库并发更新仍有上一课的陈旧回填问题，单飞并不自动解决缓存一致性。

技术文档

• Redis：分布式锁与单实例令牌
• Redis：SET
• Redis：EVAL

---

06 应对缓存雪崩：TTL 抖动、受控预热与降级

一批缓存同时到期或整个 Redis 暂时不可用，会把读请求一起推向 MySQL。把到期时间错开，并给回源设置明确上限。

从一个问题开始

一批缓存同时到期或整个 Redis 暂时不可用，会把读请求一起推向 MySQL。把到期时间错开，并给回源设置明确上限。

这次的终点

观察 45–75 秒的到期分布、最多 20 条预热；缓存停机时仍能有界读取 MySQL，恢复后重新命中缓存。

开始前，确认你在这里

• 已完成 Redis 第 01–05 课，shop-service 中商品读取脚本可用；先正常运行 cache_read.py 确认基线，再只对本课 shop-redis 做停止与恢复实验。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 新终端先 source redis-lab/local.env。第 02 课会把 MYSQL_* 和 PRODUCT_ID 一起保存在同一受限文件，后续无需重新输入密码或猜商品编号；不要重新运行初始化来生成新 Redis 密码。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 观察一批键的 TTL 分散在 45–75 秒，减少同时到期
• 顺序预热最近至多 20 件商品，给每次回源设置节奏
• 停止本机缓存后有界读取 MySQL，并验证 Redis 恢复后重新命中

一步步完成案例

1. 观察有界 TTL 抖动分布

保存为 cache_jitter.py。生成 100 个专用小键，TTL 均在 45 到 75 秒，真实设置后可能因执行消耗略减 1 秒。只清理本脚本创建的键，不扫描并删除其他缓存。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_jitter.py，再与下方预期对照。

文件：cache_jitter.py

import random
from collections import Counter
from redis_client import r

keys = [f"shop: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)

你应该看到什么

输出 min/max 均在 45–75，buckets 出现多个秒数且计数合计 100；本节临时键最终清理。

刚才发生了什么

给 TTL 加少量随机变化叫抖动，可以避免同一批键同时到期。它分散正常过期的回源压力，但不能防止整个 Redis 实例停机。

2. 用MySQL有限读取预热最近20条

保存cache_warm.py。先在短事务读取最多20个编号，commit并close后再逐个回源填充。每次load_product都重新打开短RC读取，删除期间得到None则跳过；不是一致性全库导出。顺序预热、每条50毫秒与TTL抖动限制本机负载，生产使用池及实际热点统计。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_warm.py，再与下方预期对照。

文件：cache_warm.py

import random
import time
from mysql_client import connect
from product_cache import key,load_product,encode
from redis_client import r

conn=connect()
try:
    with conn.cursor() as cur:
        cur.execute("SELECT id FROM products 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 product_id in ids:
    product=load_product(product_id)
    if product is not None:
        r.set(key(product_id),encode(product),ex=60+random.randint(-15,15))
    time.sleep(0.05)
print("warmed",len(ids))

你应该看到什么

python cache_warm.py 打印 warmed 的编号数且不超过 20；对应仍存在商品的缓存被写入不同 TTL。

刚才发生了什么

预热是用户来读之前先填一些预计会用到的数据。本例先释放取编号的数据库连接再逐件读取，避免长时间持有一笔大事务；不是全库一致性导出。

3. 实现有并发上限的回源降级

保存为 cache_fallback.py。该函数直接实现读路径，不调用可能已经回源的 get_product 后再次回源，避免重复数据库查询。Semaphore 只约束本进程；多进程预算需要总量规划。同步代码不能直接阻塞 async 路由。

文件：cache_fallback.py

import random
import threading
from redis.exceptions import RedisError
from redis_client import r
from product_cache import key, load_product, encode, decode

db_slots = threading.BoundedSemaphore(4)

class SourceBusy(RuntimeError):
    pass

def get_available(product_id, loader=load_product):
    try:
        raw = r.get(key(product_id))
        if raw is not None:
            try:
                return decode(raw, product_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:
        product = loader(product_id)
    finally:
        db_slots.release()
    if product is not None:
        try:
            r.set(key(product_id), encode(product), ex=60 + random.randint(-15, 15))
        except RedisError:
            pass  # 生产记录 redis_fill_error 计数
    return product, "database"

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。cache_fallback.py 的数据库通行证为 4；等不到 0.05 秒内可用名额会抛 SourceBusy，GET/SET 的 Redis 故障可绕过。

刚才发生了什么

降级就是某个依赖不可用时换一条有限能力的路径。MySQL 是最后的数据来源，不能因 Redis 故障就无限放大回源并发；四个名额只限制当前进程。 Redis GET/SET 失败可以绕过，MySQL 异常仍交给调用者。

4. 停掉本机缓存并验证仍可读取真源

先在终端执行 docker stop shop-redis，仅影响本课专用实例；运行下面脚本，预期来源 database。完成后 docker start shop-redis 并再运行两次，后续可恢复 cache。不要在服务正在高负载时无保护地执行全局故障演练。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_outage_probe.py，再与下方预期对照。

文件：cache_outage_probe.py

import os
from cache_fallback import get_available

product, source = get_available(int(os.environ["PRODUCT_ID"]))
assert product is not None
print(source, product["id"], product["version"])

# Shell 完成后恢复：docker start shop-redis
# 本课 Redis 未开启持久化，重启后的缓存需要重新构建。

你应该看到什么

停止 shop-redis 后脚本打印 database 与真实商品版本；恢复后重复读取由 database 转为 cache。

刚才发生了什么

Redis 读写失败不妨碍本次返回已从 MySQL 取得的数据。恢复后副本重建，缓存重新承担加速作用；MySQL 本身失败仍应作为读取失败处理。

检查学习结果

• TTL 分布有上下界，批量键没有全部同刻过期。
• 预热数量和速率有限，未把全库一次性加载。
• 专用 Redis 停止时仍读到 MySQL 数据，回源超出预算时返回受控错误。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

发布或 Redis 重启后大量缓存同时为空，数据库必须保持可用。

实现决策

• 到期抖动、限速预热与故障降级分别处理不同触发条件。
• 数据库回源有硬并发预算。
• 允许陈旧数据的范围由业务定义。
• 回源同步驱动为PyMySQL；异步API调用改用对应asyncmy接入，不在事件循环中直接执行同步SQL。

隔离环境演练

仅停止 shop-redis，使用有限并发读取并测量回源峰值，随后恢复实例。

恢复验收

• 不会无限回源或无限等待。
• 数据库错误仍可见。
• 故障恢复后命中率逐步恢复而非立即全量预热。

遇到问题时查这里

缓存断开后数据库连接数瞬间暴涨

原因：降级没有并发限制，重试与每次新建连接叠加。

处理：按进程规划回源预算，使用池和短等待，超过容量快速拒绝；避免多层重试。

抖动后出现零或负 TTL

原因：随机偏移超过基础期限。

处理：明确最小 TTL 并测试范围，例如 max(1,base+jitter)，业务过期上限也要符合要求。

自己试一次

在 Redis 停止时用 12 个线程和每次 0.2 秒的 loader 调用 get_available，预测 SourceBusy 的原因。

提示：本进程只有 4 个回源名额，等待最多 0.05 秒。

参考答案：

前 4 个请求可占用回源名额，其余请求可能在名额释放前超时并得到 SourceBusy。实际数量由调度决定，验收应看峰值回源并发不超过 4、等待有界、恢复后正常，而非硬编码某个固定失败数。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

雪崩影响一批键或整个实例

完成第 1 步“观察有界 TTL 抖动分布”后，再回顾这条规则：同一时刻大量键到期，或 Redis 整体不可用，会把原本被缓存吸收的流量推向数据库。TTL 抖动分散到期时刻，不能解决实例故障；需要回源限流和服务端负载保护。

降级也有容量上限

完成第 3 步“实现有并发上限的回源降级”后，再回顾这条规则：Redis故障不代表MySQL能接住全部流量。回源需按进程设置并发名额、池等待、驱动读写超时、InnoDB锁等待和SELECT执行预算；超量返回明确暂不可用。即使数据库默认RR，应用连接也显式RC并结束每次只读事务，避免复用连接时长期保留旧快照。

预热从热点开始

完成第 2 步“用MySQL有限读取预热最近20条”后，再回顾这条规则：预热最常访问的有限集合，并控制批次大小、并发和速率。全库一起加载会抢占生产数据库资源和缓存容量。旧值兜底只适用于允许陈旧的字段，权限、删除状态和敏感信息不能随意继续返回。

技术文档

• Redis：客户端生产使用
• Redis：EXPIRE
• Redis：内存淘汰

---

07 使用 Pipeline、WATCH 与 Lua 保证正确的原子操作

发送一批命令、发现别人改过数据、把计数和期限一起设置，是三种不同需求。用小计数实验分别观察该选哪种工具。

从一个问题开始

发送一批命令、发现别人改过数据、把计数和期限一起设置，是三种不同需求。用小计数实验分别观察该选哪种工具。

这次的终点

批量返回顺序可核对，竞争修改触发重读，限流计数前三次允许、第四次拒绝且到期恢复。

开始前，确认你在这里

• 已完成 Redis 第 01–06 课，在 shop-service 终端运行 redis_client.py 确认 PING 成功；本课只使用明确前缀的小实验键，不需要新商品数据。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 新终端先 source redis-lab/local.env。第 02 课会把 MYSQL_* 和 PRODUCT_ID 一起保存在同一受限文件，后续无需重新输入密码或猜商品编号；不要重新运行初始化来生成新 Redis 密码。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 用 Pipeline 核对批量命令返回顺序，并区分批量与原子执行
• 制造 WATCH 冲突，重新读取后将计数从 10 增加到 11
• 用 Lua 同时计数和设置期限，验证前三次允许、第四次拒绝及到期恢复

一步步完成案例

1. 批量发送并理解返回位置

保存为 redis_pipeline.py。transaction=False 只减少往返，返回列表按命令顺序排列。若一次批量中途网络断开，不能仅凭客户端异常判断所有命令都没执行。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_pipeline.py，再与下方预期对照。

文件：redis_pipeline.py

from redis_client import r

keys = [f"shop: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)

你应该看到什么

打印 [True, True, True, '0', '1', '2']，与三条 SET 再三条 GET 的顺序一致，最后清理三键。

刚才发生了什么

Pipeline 把多条命令一起发出，减少来回网络次数。transaction=False 不保证这组命令之间无人插入其他操作，批量发送与原子执行是两回事。 专用键全部带 TTL，并在验收后显式删除。

2. 制造 WATCH 冲突并重读

保存为 redis_watch.py。第一次 watch 后，使用客户端的另一次命令修改该键，确保 EXEC 抛 WatchError；第二次重新 watch 和读取，应从值 10 增加到 11。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_watch.py，再与下方预期对照。

文件：redis_watch.py

from redis import WatchError
from redis_client import r

k = "shop: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)

你应该看到什么

第一次打印 conflict, retry from read；重试后键值为 11，断言通过，不会按旧值 0 写成 1。

刚才发生了什么

WATCH 记住准备修改的键是否变过。提交前发现别人已经写成 10 就拒绝旧计算，必须重新读取再算，这叫乐观检查。 最多尝试 3 次，示例没有无限循环。

3. 实现原子计数与固定期限

保存为 redis_limiter.py。窗口从首次请求开始，不是精确滑动窗口。网络超时重试可能再次计数，业务必须允许保守限流或使用独立请求幂等方案。

文件：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"shop:lab:limit:{{{subject}}}"
    allowed, count, ttl_ms = limit_script(keys=[k], args=[3, 1000])
    return bool(allowed), count, ttl_ms

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。redis_limiter.py 中 INCR 与首次 PEXPIRE 放在同一短脚本，返回允许标志、当前计数与剩余毫秒。

刚才发生了什么

如果把计数和期限分两次发送，中途断网可能留下永不过期的计数器。Lua 让这组操作不中断地执行；先验证参数避免改到一半才发现输入不合法。 脚本验证限制与窗口后 INCR，首次计数同时 PEXPIRE，避免 INCR 成功但 EXPIRE 未发出的永久计数器。

4. 验证窗口内拒绝和到期恢复

保存为 redis_limiter_probe.py。生产映射 429，并根据 TTL 向上取整计算 Retry-After；权限鉴定接口与普通缓存故障时的放行策略必须分别设计。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_limiter_probe.py，再与下方预期对照。

文件：redis_limiter_probe.py

import time
from redis_client import r
from redis_limiter import allow_request

k = "shop: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)

你应该看到什么

四次允许结果为 True、True、True、False，第四次 TTL 为正；约 1.1 秒后再次允许，测试键被删除。

刚才发生了什么

这是从首次请求开始的固定窗口计数。被拒绝时剩余时间可以告诉客户端多久后再试；它并不保证任意滑动一秒内都至多三次。 前 3 次允许，第 4 次拒绝且返回正的剩余毫秒；约 1.1 秒后新窗口允许。

检查学习结果

• Pipeline 返回顺序与命令顺序一致，批量大小有限。
• WATCH 实验先失败再从新值重试，不覆盖外部修改。
• 限流前三次允许、第四次拒绝、窗口到期恢复，计数键存在时带 TTL。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

API 需要在多进程之间共享请求计数，并避免计数和过期配置分离。

实现决策

• 流水线、事务和 Lua 各用于明确的操作边界。
• 脚本短且输入受限。
• 限流故障策略不沿用普通数据缓存的无条件放行。

隔离环境演练

在专用键制造 WATCH 冲突与窗口超额请求，验证有限重试和原子期限。

恢复验收

• 没有永久计数键。
• 事务错误不被错误描述为自动回滚。
• 超时重试与边界突发的业务含义明确。

遇到问题时查这里

误把 transaction=False 的 Pipeline 当成原子业务操作

原因：减少网络往返与命令不可交错是不同保证。

处理：需要检查并更新用 WATCH，需要短组合操作用 Lua；明确事务选项。

INCR 后进程崩溃留下永不过期限流键

原因：EXPIRE 是另一条命令，没来得及发出。

处理：把首次计数和到期设置放进短脚本；对专用命名空间监控 TTL=-1 异常。

自己试一次

窗口边界附近为什么可能在很短时间内通过接近两倍限制的请求？

提示：上一窗口末尾和下一窗口开头分别拥有各自额度。

参考答案：

固定窗口允许边界突发：旧窗口末尾用完额度，过期后新窗口立即重新计数。需要更平滑限制时可选择滑动窗口或令牌桶，评估精度、状态量和脚本成本；不能把本例当作精确的任意 1 秒最多 3 次。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

Pipeline 首先减少网络往返

完成第 1 步“批量发送并理解返回位置”后，再回顾这条规则：transaction=False 的 pipeline 批量发送命令，不让多条业务操作自动变成一个不可分割动作。redis-py pipeline 默认启用事务，教学中明确写参数以免混淆。大批次会占内存和延长等待，应限制规模。

WATCH 是提交前的乐观检查

完成第 2 步“制造 WATCH 冲突并重读”后，再回顾这条规则：WATCH 键后读取当前值，MULTI 与 EXEC 提交修改；被其他客户端改动时 redis-py 抛 WatchError，应用重新读取并有限重试。Redis 事务与 MySQL 不同，执行阶段某条命令出错不会撤销已经成功的其他命令。

Lua 组合命令时先验证输入

完成第 3 步“实现原子计数与固定期限”后，再回顾这条规则：短 Lua 脚本在执行期间不被其他命令穿插。原子执行不代表发生运行时错误会回滚此前的写入。参数与数据类型应在变更前校验，脚本必须短，不能用长循环阻塞所有客户端。

技术文档

• Redis：Pipeline 与事务
• Redis：事务语义
• Redis：Lua 脚本
• Redis：INCR

---

08 控制缓存容量与淘汰，隔离任务 Broker

缓存空间有限，新的商品值写进来时旧副本可能被挤掉。用独立小实例观察淘汰，并解释任务消息为什么不能混用同一容量。

从一个问题开始

缓存空间有限，新的商品值写进来时旧副本可能被挤掉。用独立小实例观察淘汰，并解释任务消息为什么不能混用同一容量。

这次的终点

8MB 独立实例产生可测的淘汰增量；区分命中率、到期与淘汰，并写出缓存与 broker 的独立容量安排。

开始前，确认你在这里

• 已完成 Redis 第 01–07 课，redis-lab/redis.conf 和 REDIS_PASSWORD 可用；在 shop-service Bash 确认 6381 与 shop-redis-capacity 属于本节可新建的环境。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 新终端先 source redis-lab/local.env。第 02 课会把 MYSQL_* 和 PRODUCT_ID 一起保存在同一受限文件，后续无需重新输入密码或猜商品编号；不要重新运行初始化来生成新 Redis 密码。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 在独立 8MB 实例有限写入，记录本次淘汰增量与内存占用
• 统计指定 2000 个键的命中率，区分未命中、过期和淘汰
• 为可重建缓存与任务 broker 制定独立容量、保留和拒写策略

一步步完成案例

1. 启动单独容量实验实例

以下 Linux Bash 只创建 shop-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 shop-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

你应该看到什么

新容器只映射 6381，capacity.conf 的 maxmemory 为 8mb；原 shop-redis 仍使用 6380 和 128mb 配置。

刚才发生了什么

单独小容量实例让淘汰实验不影响正在使用的商品缓存。maxmemory 限制 Redis 管理的数据内存，进程实际占用还包含其他开销。

2. 有限写入制造容量压力

保存为 redis_capacity.py。运行前核对端口 6381。记录写入前后 evicted_keys 差值，不能用实例启动以来累计值直接推算本次淘汰。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_capacity.py，再与下方预期对照。

文件：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"shop: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()

你应该看到什么

打印 evicted delta 大于 0，remaining keys 少于 2000，policy 为 allkeys-lru；同时可见 used/rss/max 三个不同数字。

刚才发生了什么

未到十分钟 TTL 的键也可能因容量不足被淘汰。LRU 策略倾向移除较少近期使用的键；对比前后计数才知道本次实验淘汰多少。 最多写 2,000 个约 8KB 字符串，带 10 分钟 TTL；请求数和内存实验都有上限。

3. 区分命中率与淘汰率

在容量实例读取 2,000 个明确键，统计本次命中和未命中。不要用 KEYS * 在大实例全量扫描。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_capacity_hits.py，再与下方预期对照。

文件：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"shop: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()

你应该看到什么

打印本批 hits、misses 且相加为 2000，ratio 为 0–1 范围；INFO 同时提供累计命中和淘汰指标。

刚才发生了什么

命中率衡量这一批读取找到多少值，淘汰数衡量内存挤掉多少键。两者相关但不相等，还需要区分自然过期造成的缺失。 下面本地计数是这批请求的真实比例；INFO 的 keyspace_hits/misses 是实例全局累计值，做监控时用区间增量。

4. 形成缓存与 Broker 的独立配置

把下列配置记录加入部署设计。容量实验完成后执行 docker stop shop-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 或故障隔离

你应该看到什么

部署记录分别写出 cache、broker、结果缓存的保留和拒写策略；完成后只停止 shop-redis-capacity。

刚才发生了什么

商品副本可回源重建，任务消息被淘汰可能导致工作丢失。不同 Redis DB 仍共用内存和进程，因此隔离可靠性需要独立实例与容量预算。

检查学习结果

• 专用容量实例 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 编号只能减少键名碰撞。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

maxmemory 不是进程总内存上限

完成第 1 步“启动单独容量实验实例”后，再回顾这条规则：Redis 的 maxmemory 控制参与淘汰计算的内存范围，进程 RSS、复制缓冲、持久化重写等还需要额外预算。容器内存上限应留余量，不能把 maxmemory 与容器限制设成完全相同。

淘汰与到期不同

完成第 2 步“有限写入制造容量压力”后，再回顾这条规则：到期按 TTL 删除，淘汰在内存压力下依据策略移除键。allkeys-lru 可从所有键中近似选择较久未使用的键；noeviction 在不足时拒绝某些需要新增内存的写命令。没有任何策略保证缓存永不丢失。

缓存实例不承载任务可靠性

完成第 4 步“形成缓存与 Broker 的独立配置”后，再回顾这条规则：任务 Broker、结果存储、锁与缓存可能要求不同持久化、淘汰和资源策略。逻辑 DB 0 与 DB 1 仍共享同一进程内存、淘汰配置和故障域。队列数据应按任务系统要求使用独立实例，noeviction 也只是拒写，不会自动让队列可靠。

技术文档

• Redis：键淘汰
• Redis：INFO
• Redis：内存优化

---

09 配置连接池、超时与熔断并记录缓存指标

Redis 连续失败时，每个商品请求都重新等一次超时会放大延迟。限制借连接和访问的等待，再短暂跳过已知失败的 Redis。

从一个问题开始

Redis 连续失败时，每个商品请求都重新等一次超时会放大延迟。限制借连接和访问的等待，再短暂跳过已知失败的 Redis。

这次的终点

能够区分正常命中、正常未命中、Redis 错误和主动绕过；连续失败后短暂跳过，冷却后一次探测恢复。

开始前，确认你在这里

• 已完成 Redis 第 06–08 课，理解有界回源；在 shop-service 保留 PRODUCT_ID、product_cache.py 与 redis_client.py，先确认 6380 缓存正常。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 新终端先 source redis-lab/local.env。第 02 课会把 MYSQL_* 和 PRODUCT_ID 一起保存在同一受限文件，后续无需重新输入密码或猜商品编号；不要重新运行初始化来生成新 Redis 密码。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 分别限制连接池等待、建立连接和访问 Redis 的时间
• 验证三次失败后主动绕过，冷却探测成功后恢复访问
• 在输出中区分命中、未命中、Redis 错误与 bypass，并记录实际延迟

一步步完成案例

1. 显式配置池与关闭顺序

保存为 redis_pool_client.py。BlockingConnectionPool 让借不到连接时最多等 50 毫秒；connect/socket 各 100 毫秒，Retry(...,0) 禁止本层自动重试，避免示例将预算偷偷翻倍。同步客户端供同步代码使用，async 路由选择 redis.asyncio 并 await。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_pool_client.py，再与下方预期对照。

文件：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()

你应该看到什么

python redis_pool_client.py 正常打印 True 并关闭池；最多 8 条连接，借连接等待 0.05 秒，连接与读写各 0.1 秒。

刚才发生了什么

连接池复用连接以减少每次建立连接的成本。池等待、建立连接和访问分别占用时间；禁用本层自动重试避免一次故障悄悄花两份预算。

2. 实现线程安全的短期熔断器

保存为 redis_breaker.py。只包裹 Redis 操作；业务 SQL 和 JSON 解析错误不能自动判作 Redis 断网。AUTH 等配置错误应报警修配置，不用无限探测掩盖。

文件：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"

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。RedisBreaker 类保留失败数、冷却截止、探测标志和 generation；仅连接/超时失败进入 redis-error，认证错误向外报出。

刚才发生了什么

熔断器在连续失败后暂时跳过依赖，冷却后允许探测一次。锁保护共享状态，generation 防止旧请求晚返回后覆盖新一轮判断；SQL 或坏 JSON 不能归为 Redis 断网。 调用方对返回的 bypass 标志进入第 06 课的有界回源流程。

3. 不用停服务验证状态转移

保存为 redis_breaker_probe.py。注入固定抛错函数，前 3 次真实失败，第 4 次直接 bypass；冷却后成功探测，下一次正常执行。此脚本先验证顺序行为；实现中的 generation 用于丢弃旧一代在途请求对新状态的覆盖，并发测试还需验证半开阶段只启动一个探测。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_breaker_probe.py，再与下方预期对照。

文件：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")

你应该看到什么

前三次打印 redis-error，第四次 bypass，broken 只被调用 3 次；冷却后打印 recovered，后续恢复正常执行。

刚才发生了什么

bypass 表示这次主动没发请求，并不等于正常访问但键不存在。先用固定故障函数验证状态转换，就不用靠停服务碰运气才能复现。

4. 记录实际调用状态与延迟

保存为 cache_metrics_probe.py。故障实验只停止 shop-redis 后运行，再重启后运行。下面状态为教学输出，生产导出聚合 counter/histogram，不记录密码、完整值或每条商品作为标签。Redis miss 是 value=None 且 status=redis，不能与 bypass 合并成命中。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_metrics_probe.py，再与下方预期对照。

文件：cache_metrics_probe.py

import os
import time
from product_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["PRODUCT_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()

你应该看到什么

缓存停机时输出 redis-error 后出现 bypass；恢复后新运行可见 status=redis，hit 取决于当前键是否存在，每行有 duration_ms。

刚才发生了什么

状态说明走了哪条路径，延迟说明付出多少等待。只有 status=redis 且 value 非空才叫命中，绕过与未命中都不能计入成功缓存读取。

检查学习结果

• 池等待、连接、socket 超时与重试次数均显式设置。
• 失败阈值、bypass、冷却探测与恢复能通过注入实验验证。
• 指标区分命中、未命中、Redis 错误和主动绕过，不记录高基数或敏感数据。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

Redis 网络故障时，API 需要在可接受延迟内切换到受控回源，并在恢复后重新命中。

实现决策

• 超时预算以完整请求为单位计算。
• 进程内熔断只负责快速跳过 Redis，回源仍受容量限制。
• 通过状态代次阻止在途旧响应覆盖新状态，并对半开探测做并发验收。

隔离环境演练

注入连续连接错误验证熔断转移，再用专用实例停止恢复观察实际延迟和来源。

恢复验收

• 故障延迟有上限。
• 没有把所有 Redis 错误当成普通 miss。
• 连接池在进程退出时关闭。

遇到问题时查这里

Redis 已故障，但每个请求仍等很久

原因：多层超时和自动重试叠加，没有总预算。

处理：画出一次请求的所有等待预算，限制重试所在层；以实测故障延迟验收。

一个并发成功请求把刚打开的熔断立即关闭

原因：多个已在途调用同时回写熔断状态，若删除 generation 检查就会出现旧请求覆盖新状态。

处理：保留 generation 校验：打开熔断或进入半开时推进代次，旧代次结果不改变新状态；补充并发探测测试或使用已有成熟组件。

自己试一次

为什么缓存 GET 超时可以考虑回源，而任务入队超时不能直接断言“任务没入队”？

提示：响应丢失不等于服务器没执行命令。

参考答案：

GET 是读取，失败后可按预算回源；入队命令可能已执行但响应丢失，直接再次入队可能重复。使用任务 ID、幂等消费、投递记录与状态核对，重试策略按操作语义设计。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

超时需要放进请求总预算

完成第 1 步“显式配置池与关闭顺序”后，再回顾这条规则：连接超时、socket 超时、池等待和重试时间会叠加。API 总预算只有一秒时，不能给每层都设置一秒再重试三遍。写命令超时可能已经执行，重试 INCR、队列入队等操作需要额外语义设计。

熔断短期跳过已知失败依赖

完成第 2 步“实现线程安全的短期熔断器”后，再回顾这条规则：连续 Redis 连接错误达到阈值后短暂打开熔断；冷却后只允许一个探测请求，其余请求继续走受控回源。恢复探测成功才关闭。下面是线程安全的进程内教学实现，不是跨所有进程同步的全局开关。

命中率不是唯一目标

完成第 4 步“记录实际调用状态与延迟”后，再回顾这条规则：高命中率也可能返回陈旧数据，低命中率也可能是新内容正常访问。至少同时观察缓存调用延迟、连接错误、回源并发、数据库延迟、失效失败与版本新鲜度。计数标签不能使用每个 product_id，否则监控基数失控。

技术文档

• Redis：生产客户端使用
• Redis：Python 错误处理
• redis-py 5.3：连接池与重试参数

---

10 演练重启、复制与持久化恢复

缓存重启后可以从 MySQL 重建，但有些数据希望重启仍在。分别验证临时缓存、保存写入记录的实例和只读副本。

从一个问题开始

缓存重启后可以从 MySQL 重建，但有些数据希望重启仍在。分别验证临时缓存、保存写入记录的实例和只读副本。

这次的终点

正常重启后缓存可重建，AOF 样本可恢复，副本能读取主节点数据；主节点停止时副本不会自动升主。

开始前，确认你在这里

• 已完成 Redis 第 01–09 课；shop-service 的 PRODUCT_ID 有效，6380 正常；本节另用 6382/6383 与专用卷/网络，先核对容器名端口没有被其他项目占用。
• 所有 Redis 脚本都放在 Linux/WSL2 的 shop-service 根目录，与 mysql_client.py、redis_client.py 同级，不放进 tasks_lab。每个终端先 cd 到该目录、source .venv/bin/activate；本页 Shell 代码使用 Bash，Windows 不要粘贴进 PowerShell。
• 新终端先 source redis-lab/local.env。第 02 课会把 MYSQL_* 和 PRODUCT_ID 一起保存在同一受限文件，后续无需重新输入密码或猜商品编号；不要重新运行初始化来生成新 Redis 密码。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 重启无持久化缓存，验证商品仍可从 MySQL 读取并回填
• 正常重启独立 AOF 实例后只读验证样本值和剩余期限
• 确认主节点写入传到副本，并观察主节点停止时副本不会自动升主

一步步完成案例

1. 确认无持久化缓存重启后可回源

使用第 02 课 PRODUCT_ID 预热后，记录缓存存在。执行 docker restart shop-redis，仅重启专用本机实例，然后再次读取应先 database 再 cache。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python cache_restart_probe.py，再与下方预期对照。

文件：cache_restart_probe.py

import os
from product_cache import get_product, key
from redis_client import r

n = int(os.environ["PRODUCT_ID"])
print(get_product(n)[1])
print("cache exists", bool(r.exists(key(n))))
# 终端执行 docker restart shop-redis 后，再运行：
# print(get_product(n)[1])  -> database
# print(get_product(n)[1])  -> cache

你应该看到什么

重启前 cache exists 为 True；重启 shop-redis 后第一次来源 database，紧接着为 cache，MySQL 商品仍存在。

刚才发生了什么

本机主缓存关闭了落盘保存，重启会清空副本。缓存失去内容不等于商品丢失，因为商品仍以 MySQL 为准。 配置 save "" 与 appendonly no 禁用了持久化，但 MySQL 的商品仍然存在。

2. 创建独立 AOF 实例

以下 Linux Bash 创建 shop-redis-aof，端口 6382，命名卷 shop_redis_aof 保存 /data。正式镜像应记录具体 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 shop-redis-aof -d -p 127.0.0.1:6382:6379 \
  -v "$PWD/redis-lab/aof.conf:/usr/local/etc/redis/redis.conf:ro" \
  -v shop_redis_aof:/data \
  redis:7 redis-server /usr/local/etc/redis/redis.conf

你应该看到什么

shop-redis-aof 映射 6382，配置 appendonly yes / everysec，数据写入独立命名卷 shop_redis_aof。

刚才发生了什么

AOF 记录写入操作，重启时用它重建数据；everysec 表示刷盘策略，不能据此保证任意崩溃时最后一笔都保存。命名卷提供容器之外的持续存储位置。 由已有配置生成 AOF 配置，主业务缓存 6380 保持原样。

3. 验证 AOF 状态与正常重启恢复

保存为 redis_aof_probe.py。第一次运行写入样本并输出 aof_enabled、aof_last_write_status；终端 docker restart shop-redis-aof 后仅执行读取部分，不再次写入掩盖丢失。样本 TTL 10 分钟，重启后剩余时间减少；超过期限消失符合语义。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python redis_aof_probe.py，再与下方预期对照。

文件：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 = "shop: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()

你应该看到什么

首次可见 aof_enabled=1、aof_last_write_status=ok；重启后只读取仍得到 persist-me，TTL 比之前减少。

刚才发生了什么

必须只读验证恢复，否则重新 SET 会把丢失掩盖。正常重启恢复说明这条样本被保留，不代表已经模拟了所有断电、磁盘和备份故障。

4. 建立只读副本并检查复制

为 6382 的 AOF 主实例创建本机 Docker 网络，添加无持久化副本 shop-redis-replica，宿主机 6383。配置中的 masterauth 使用同一练习密码；生产应采用专用 ACL 凭据和加密通道。副本首次同步可能需要等待，INFO replication 中 master_link_status=up 才进入验收。

文件：终端 / 当前课时

docker network create shop-redis-lab
docker network connect shop-redis-lab shop-redis-aof
python3 - <<'PY'
import os
from pathlib import Path
source = Path('redis-lab/redis.conf').read_text()
source += 'replicaof shop-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 shop-redis-replica --network shop-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

你应该看到什么

副本启动后 INFO replication 的 master_link_status 最终为 up，角色为 replica/slave；在同步完成后再运行下一步。

刚才发生了什么

复制把主节点写入传给另一实例，副本只读能防止随意形成两套数据。链路 up 说明连接建立，后面仍要用具体新写入验证数据到达。

5. 核对复制、主节点故障和恢复边界

在 shop-service 保存 redis_replica_probe.py。先确认副本链路 up，再运行 python redis_replica_probe.py 并观察副本确认数与读取值。随后只停止专用 shop-redis-aof，查询副本 INFO replication，核对链路 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("shop: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("shop:lab:replica-probe"))
finally:
    primary.close()
    replica.close()

# Linux 故障操作仅针对专用实例：docker stop shop-redis-aof
# 检查副本 INFO replication；不会自动升主。
# 恢复：docker start shop-redis-aof
# 演练结束：docker stop shop-redis-aof shop-redis-replica

你应该看到什么

WAIT 返回确认数至少 1，副本读取 replicated；停止主节点后副本链路转 down 且仍未自动升主，恢复主节点可重新连接。

刚才发生了什么

WAIT 在写入的同一连接上等副本确认，不等同于磁盘持久化或自动故障切换。复制、恢复数据和选新主节点是不同能力，本节只验证前两者的明确边界。 pipeline(transaction=False) 在本例保持 SET 与 WAIT 使用同一连接，并按命令顺序返回 written 和 acknowledged；若改成两条可能借不同连接的独立调用，就无法按同一写入连接验证等待。

检查学习结果

• 6380 重启清空缓存后，业务商品能从 MySQL 读取并重建。
• 6382 AOF 正常重启后样本仍可读取，TTL 正常减少，持久化状态无错误。
• 副本连接正常时能读取复制值，主节点停止后不误认为已实现自动故障转移。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

缓存节点需要维护重启，团队必须知道数据会丢在哪里、服务如何继续以及恢复验收是什么。

实现决策

• 按数据用途决定是否持久化。
• 正常重启、异常断电与主从切换分开测试。
• 复制不是数据库真源和任务幂等性的替代。

隔离环境演练

仅重启或停止 6380/6382/6383 这些本机专用实例，分别记录回源、AOF 和复制状态，再恢复或停止实验环境。

恢复验收

• 真实商品不丢失。
• 没有宣称 everysec 零丢失。
• 未部署自动选主时明确记录为手动恢复实验。
• Redis重启只影响可重建副本；MySQL中的商品、业务version与UTC时间保持可核对。

遇到问题时查这里

启动了副本就宣称零丢失高可用

原因：忽略异步复制窗口、自动选主与客户端切换。

处理：分别验证复制延迟、故障转移组件和客户端发现；关键业务状态保存在适合其可靠性要求的真源。

重启恢复测试总成功，实际上每次都重新 SET

原因：验收脚本在读取之前又写入样本，掩盖持久化失败。

处理：将写入阶段与重启后只读验收分开，记录初始值、时间、TTL 和持久化状态。

副本长期显示 master_link_status=down

原因：Docker 网络、主实例名称或 masterauth 不匹配。

处理：检查主实例是否加入 shop-redis-lab、名称是否一致，以及双方认证配置；不要通过关掉认证规避问题。

自己试一次

为缓存实例故障写一份最短恢复顺序，并说明是否必须先恢复所有旧缓存键。

提示：先保证真源和受控回源，再逐步恢复缓存。

参考答案：

确认 MySQL 正常，启用回源限流；检查 Redis 实例身份、认证和配置；恢复连接后小批次预热热点并观察错误率与数据库延迟。可重建缓存不必恢复所有旧键。任务 Broker 需要额外检查积压、未确认任务和重复消费，不能沿用此简化流程。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

RDB 与 AOF 保存不同形式

完成第 2 步“创建独立 AOF 实例”后，再回顾这条规则：RDB 保存某个时点的快照；AOF 记录写操作并按 fsync 策略落盘。appendfsync everysec 在故障下仍有近期数据丢失窗口。正常重启看到值还在，只证明该次正常路径，不能据此宣称断电零丢失。

复制不是自动故障转移

完成第 5 步“核对复制、主节点故障和恢复边界”后，再回顾这条规则：Redis 主从默认异步复制，主节点确认写入时副本可能尚未收到。WAIT 可等待副本确认复制，仍不把系统变成强一致存储。生产自动故障转移由 Sentinel、Cluster 或受管平台等承担，部署拓扑与客户端发现都需独立验证。

恢复验收从真源开始

完成第 1 步“确认无持久化缓存重启后可回源”后，再回顾这条规则：先确认MySQL 8.4业务数据完整、InnoDB可读写，再检查Redis角色、复制与受控回源。Redis AOF、RDB与MySQL的redo、binlog是不同日志，不能互相替代。缓存可丢时可从MySQL重建；任务Broker还需验收持久Outbox、未确认任务与幂等消费，不用缓存重启结果证明任务可靠性。

技术文档

• Redis：持久化
• Redis：复制
• Redis：WAIT
• Redis：Sentinel