在应用程序中使用 Redis

把 Redis 装好、在命令行里跑通 SET 和 GET，只是第一步。真正把它接进应用时，麻烦通常不在命令本身，而在命令周围：连接由谁创建，缓存失效后谁回源，一次超时能不能重试，计数会不会重复，会话该在什么时候过期，Redis 暂时不可用时接口又该怎么回答。

你可以把 Redis 想成服务旁边的一张高速工作台。正在处理的东西放上去，拿取很快；但工作台空间有限，也可能临时够不着。哪些数据可以只放在工作台上，哪些必须在仓库里留底，工作台乱了以后怎样恢复，都要由应用说清楚。只会调用命令，却没有这些边界，Redis 越快，错误传播得也可能越快。

这一章我们沿着一条真实的接入路径往前走。示例统一使用异步版 redis-py，假设 Redis 服务于一个有商品查询、下单事件、近期活动和登录会话的 Python Web 应用。你会看到的不只是“怎么写”，还有每种写法在什么条件下会翻车。

先把客户端接稳

一个进程共享一个客户端

第一次写 Redis 接入时，很容易把客户端创建放进请求函数：请求来了就连接，用完马上关闭。代码看起来很规整，实际却绕过了连接池。每个请求都要重复建立 TCP 连接、认证和协商，流量一上来，应用和 Redis 都会忙着做没有业务价值的连接工作。

redis.asyncio.Redis 客户端内部已经管理连接池。长生命周期服务更合适的做法是：应用启动时创建一个客户端，在请求和后台任务之间共享它，应用关闭时再调用 aclose()。共享的是客户端；pipeline、Pub/Sub 这类带状态的对象仍应由每个任务单独创建。

应用启动时创建共享客户端，请求借还连接，配置超时与指数退避，并在应用关闭时释放连接池。

下面先搭出一个最小而完整的生命周期。连接地址来自环境变量，密码不写进源码；启动时用 PING 做就绪检查，关停时释放池中的连接。

import os
from contextlib import asynccontextmanager
 
import redis.asyncio as redis
from fastapi import FastAPI, Request
from redis.backoff import ExponentialWithJitterBackoff
from redis.exceptions import AuthenticationError, ConnectionError, TimeoutError
from redis.retry import Retry
 
 
def build_redis_client() -> redis.Redis:
    redis_url = os.environ["REDIS_URL"]
 
    no_command_retry = Retry(
        ExponentialWithJitterBackoff(base=0.05, cap=0.5),
        retries=0,
        supported_errors=(ConnectionError, TimeoutError),
    )
 
    return redis.Redis.from_url(
        redis_url,
        decode_responses=True,
        encoding="utf-8",
        socket_connect_timeout=1.0,
        socket_timeout=0.8,
        health_check_interval=30,
        max_connections=40,
        retry=no_command_retry,
        client_name="catalog-api",
    )
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    client = build_redis_client()
    await client.ping()
    app.state.redis = client
    try:
        yield
    finally:
        await client.aclose()
 
 
app = FastAPI(lifespan=lifespan)
 
 
def get_redis(request: Request) -> redis.Redis:
    return request.app.state.redis

这里的数值只是展示配置位置，不是可以复制到所有系统的“标准答案”。连接池大小受单进程并发 Redis 操作数、命令耗时和 Redis 允许的连接数共同约束。池太小，请求会等待或拿不到连接；池太大，也不会让单条命令变快，反而可能把压力更猛烈地推给 Redis。

示例还主动把命令重试设为 0，因为这个客户端既读也写，我们不希望一次依赖升级悄悄改变写命令的重放次数。若你把只读缓存操作隔离在另一个客户端上，可以把同一个 Retry 的 retries 调成 2 或其他经过预算验证的数值；每次等待使用指数退避加随机抖动，避免大量进程同一时刻再次冲向 Redis。这个只读客户端拥有自己的连接池，也要随应用一起关闭。

超时要从接口预算倒推

连接超时解决的是“多久连不上就放弃”，命令超时解决的是“连接建立后，多久收不到结果就放弃”。如果一个接口总预算只有 300 毫秒，却让 Redis 命令等数秒，降级逻辑就算写了也来不及执行。

我们通常从用户请求的总预算往回分：网关、应用排队、主数据库、Redis 和响应序列化各自占多少。超时过长会拖住任务和连接；过短则会把本来能完成的命令取消掉，还可能触发额外重试。阻塞命令还要有单独预算，不要沿用普通 GET 的参数。

如果业务层还需要更严格的截止时间，Python 3.11 可以用 asyncio.timeout() 包住调用：

import asyncio
 
 
async def read_with_deadline(client: redis.Redis, key: str) -> str | None:
    try:
        async with asyncio.timeout(0.2):
            return await client.get(key)
    except asyncio.TimeoutError:
        return None

这里要分清两个同名异常：redis.exceptions.TimeoutError 表示客户端的套接字超时，asyncio.TimeoutError 或 TimeoutError 表示协程截止时间。工程里最好使用明确别名，日志里也把两者分开统计。

字符串解码不是无条件开启

Redis 在线路上传的是字节。decode_responses=True 会按指定编码把响应转成 Python 字符串，存 JSON、会话字段和普通计数时很省事；但如果你存的是压缩包、图片、序列化二进制或加密后的密文，就应该保留字节响应，甚至给这类用途单独建客户端。

不要一边依赖自动解码，一边在业务代码里到处写 .decode()。这种混搭在正常路径上可能没事，一遇到空值或已经是字符串的返回值就会报错。更稳的做法是按用途约定客户端：文本客户端返回 str，二进制客户端返回 bytes，序列化和反序列化集中在仓储层。

缓存要先说清谁是真相

缓存旁路的直觉

商品详情很适合缓存：读得多，改得少，偶尔短暂陈旧通常可以接受。最常见的缓存旁路流程是由应用自己管理缓存：读取时先看 Redis，未命中再查数据库并回填；写入时先提交数据库，再删除缓存。下一次读取发现未命中，就会拿到数据库里的新值。

读取时先查缓存，未命中则回源并回填；写入时先更新主数据库，再删除缓存。缓存不是事实源。

这套流程简单，但它不是强一致协议。数据库已经更新、缓存还没删掉的短暂时间里，读请求仍可能拿到旧值；删除失败时，旧值会一直留到 TTL 到期。TTL 因而不是随手填的数字，而是业务允许陈旧多久的上限之一。

先实现读取路径。下面用一个短期哨兵缓存“商品不存在”，避免大量请求持续查询同一个不存在的编号；正常商品采用较长 TTL，并加随机抖动。

import json
import secrets
from dataclasses import asdict
 
from redis.exceptions import AuthenticationError, ConnectionError, TimeoutError
 
 
NOT_FOUND = "__not_found__"
 
 
def product_cache_key(product_id: int) -> str:
    return f"prod:catalog:product:{product_id}:v3"
 
 
async def get_product(client, repository, product_id: int

注意这里把“没连上缓存”和“缓存未命中”都导向数据库，但它们必须在指标里分开。未命中是正常业务状态；连接错误是依赖故障。如果监控只看最终接口成功率，Redis 坏掉后数据库被悄悄打满，你会很晚才发现问题。

写路径先落库，再失效缓存

async def update_product(client, repository, product_id: int, patch: dict):
    # 商品更新与“待删除哪个缓存”的 outbox 记录在同一数据库事务提交
    updated = await repository.update_product_and_record_invalidation(
        product_id,
        patch,
    )
 
    try:
        await client.delete(product_cache_key(product_id))
    except AuthenticationError:
        raise
    except (ConnectionError, TimeoutError):
        # 主数据已经提交，不能盲目重做数据库写入；outbox worker 会补删
        pass
 
    return updated

为什么不直接把新值写进缓存？删除更不容易把“字段转换遗漏”“旧版本结构”“另一个写入刚刚覆盖”等问题固化进缓存。它的代价是下一次读必然回源一次。这里的 outbox 记录必须和商品更新使用同一数据库事务，后台 worker 即使重复删除缓存也没有副作用。如果读路径非常敏感，也可以写库后更新缓存，但要承担双写顺序、失败补偿和并发覆盖的复杂度。

三类击穿要分开处理

“缓存没挡住流量”背后常有三种不同原因：

• 穿透：请求的是不存在的数据，每次都回源。可以缓存短 TTL 的不存在哨兵，并在入口校验编号格式。
• 击穿：一个热门键到期，大量请求同时回源。可以做进程内 single-flight，或用带所有权令牌和过期时间的分布式加载锁。
• 雪崩：许多键在同一时刻到期，数据库突然接住整批流量。可以给 TTL 加随机抖动、分批预热，并为回源路径设置并发上限。

分布式加载锁不是越长越安全。锁的 TTL 应略长于合理的回源上界；释放时必须核对所有权令牌，不能直接 DEL，否则旧持有者可能删掉已经被别人重新取得的锁。锁等待者也不能无限轮询，应有截止时间，并准备返回旧值、降级结果或忙碌响应。

键名和 TTL 是数据模型的一部分

键名是给程序和运维一起看的

Redis 没有自动命名空间，冒号只是约定俗成的分隔符。一个实用的键通常能回答环境、服务、实体、标识和版本，例如：

prod:catalog:product:1042:v3
prod:checkout:{order-8848}:idempotency
prod:session:web:9f2c...  # 这里是随机会话号，不是用户邮箱

版本段能让结构升级更从容。旧值可以等 TTL 自然淘汰，新代码只读新版本，不必在上线瞬间扫描并改写整个键空间。对于 Redis Cluster，需要让一次多键原子操作落在同一槽时，可以把共同部分放进 {} 哈希标签；但不要为了方便把所有业务键都塞进一个标签，那会制造热点。

键名会出现在监控、慢日志、内存分析和排障截图里，所以不要把邮箱、手机号、访问令牌、完整查询文本塞进去。键太长也会消耗内存。更合适的办法是使用内部 ID 或不可逆摘要，并在应用层保留必要映射。

键名应清晰表达业务边界且避免敏感信息；缓存过期时间加入随机抖动，可以减少同时失效导致的流量尖峰。

TTL 同时表达产品语义和资源边界

给 TTL 时先问两个问题：数据多久后算旧，以及这个键如果永远不删会怎样。缓存 TTL 约束陈旧窗口；会话 TTL 表示失去活跃后的登录寿命；幂等键 TTL 则表示系统承诺识别重复请求的时间窗。三者不能因为“都是过期时间”就统一填成一小时。

同一批键使用完全相同的 TTL，会一起过期。抖动的做法是在基础 TTL 上加一个受控随机量：

import secrets
 
 
def ttl_with_jitter(base_seconds: int, jitter_seconds: int) -> int:
    if base_seconds <= 0 or jitter_seconds < 0:
        raise ValueError("TTL 参数必须是正的基础时间和非负抖动")
    return base_seconds + secrets.randbelow(jitter_seconds + 1)

抖动只是在时间上错峰，不会修复错误的业务 TTL。权限缓存如果只能容忍 10 秒陈旧，不能为了命中率把基础 TTL 拉到 10 分钟；相反，一个每晚才更新的公开字典也没必要 30 秒过期一次。

把永久键当成主动决定

不是所有键都必须过期，但“没有 TTL”应当是设计结论，不是忘记调用 EXPIRE。持久配置、受控索引可能需要永久存在；缓存、临时锁、会话、幂等标记、时间窗口计数大多应该有明确寿命。

上线后可以抽样检查：缓存键是否存在无 TTL 的异常，TTL 分布是否扎堆，过期后数据库回源量是否出现尖峰。不要用 KEYS * 在生产环境做盘点，应该使用 SCAN 渐进遍历，或者从业务写入侧直接记录 TTL 合规指标。

把业务意图映射到数据结构

Redis 的数据结构不是“哪种更高级”的选择题。先说业务要保证什么，再挑能直接表达它的命令。下面看四个常见的小模式：幂等计数、近期记录、日志统计和会话。

四种业务意图在 Redis 中的实现路径：SET NX 判重后计数、列表保留最近 N 条、流追加并限制长度、哈希会话配合滑动过期。

幂等地记录事件，再做计数

假设消息队列可能重复投递“订单已完成”事件。直接 INCR 会把重复消息算多次。我们需要一个事件号作为门票：第一次出现时占位并计数，后续重复事件只返回“已经处理”。这两个动作必须在同一个原子边界里，否则进程可能在占位后、计数前退出。

下面用 Lua 把判重、计数和两个 TTL 合并成一次服务端原子执行：

from datetime import UTC, datetime
 
 
COUNT_ONCE_SCRIPT = """
if redis.call('SET', KEYS[1], '1', 'NX', 'EX', ARGV[1]) then
  local count = redis.call('INCR', KEYS[2])
  if count == 1 then
    redis.call('EXPIRE', KEYS[2], ARGV[2])
  end
  return {1, count}
end
 
local current = tonumber(redis.call('GET', KEYS[2]) or '0')
return {0, current}
"""
 
 
async def count_checkout_once(client, event_id: str) -> tuple[bool, int]:
    day = datetime.now(UTC).strftime(

{checkout:日期} 让两个键在 Cluster 中进入同一哈希槽，代价是当天的这类操作集中到同一槽。吞吐继续增大时，可以按租户、区域或稳定分片号拆槽，再在报表层聚合。这里还要确认事件 ID 的来源确实稳定；如果重试时每次都生成新 ID，幂等键写得再漂亮也没有用。

用封顶列表保存近期记录

后台页面只想显示某个用户最近 20 次操作时，List 很合适：LPUSH 把最新记录放到头部，LTRIM 丢掉范围外的旧记录。写入和裁剪放在同一事务 pipeline 中，避免进程在两步之间退出后列表持续长大。

import json
from datetime import UTC, datetime
 
 
async def remember_recent_action(
    client,
    user_id: int,
    action: str,
    object_id: str,
) -> None:
    key = f"prod:activity:user:{user_id}:recent"
    item = json.dumps(
        {
            "action": action,

这个列表适合“最近看过什么”，不适合审计。LTRIM 会主动丢历史，Redis 的内存淘汰也可能影响数据。需要长期留存、检索和合规证明的操作日志，应进入专门的日志或审计系统。

日志流和日志统计分开保存

日志内容与统计指标解决不同问题。近期错误事件可以写入 Stream，保留顺序、字段和事件 ID；错误类型的小时计数则可以用 Hash。Stream 也要设上限，否则“临时日志”最终会变成一块没有边界的内存仓库。

import time
 
 
async def record_checkout_error(
    client,
    error_code: str,
    request_id: str,
    elapsed_ms: int,
) -> None:
    stream_key = "prod:log:checkout:errors"
    bucket = time.strftime("%Y%m%d%H", time.gmtime())
    stats_key = f"prod:metric:checkout:errors:{bucket}"
 
    async

这里用 transaction=False 是因为两个结果允许极少量偏差，目标只是减少网络往返。若统计必须和事件严格一致，就要改成事务、Lua，或干脆由同一条持久事件流异步汇总。

日志字段还要克制。请求 ID、错误码、耗时通常足够定位类别；密码、令牌、完整 Cookie、支付资料和缓存值不要写入。Redis 可以帮助排障，但它不是绕开日志脱敏规则的捷径。

会话需要完整的安全生命周期

多实例 Web 服务不能把会话只放在单机内存里，否则请求换到另一台实例就“失忆”。Redis Hash 可以按字段保存小型会话状态，TTL 自动清理不活跃会话。浏览器只持有随机、不可猜的会话 ID，真正的数据留在服务端。

import secrets
from datetime import UTC, datetime
from redis.exceptions import WatchError
 
 
SESSION_TTL = 30 * 60
 
 
def session_key(session_id: str) -> str:
    return f"prod:session:web:{session_id}"
 
 
async def create_session(client, user_id: int) -> str:
    session_id 

这是教学版骨架。真实系统还应在登录成功、权限提升和敏感资料修改后轮换会话 ID；Cookie 设置 HttpOnly、Secure 和合适的 SameSite；退出时删除服务端会话并清除 Cookie。会话里只放请求频繁使用的小字段，用户完整资料仍留在主数据库。

会话对故障的处理也不同于商品缓存。缓存读失败可以回源数据库；会话读失败时，随意放行会绕过身份状态，通常应失败关闭或返回暂时不可用。所谓“优雅降级”，前提是没有降低安全边界。

Pipeline、事务和乐观锁别混在一起

Pipeline 优化的是往返

客户端逐条执行 100 个独立命令，通常要经历许多次网络往返。Pipeline 允许先在客户端排队，再成批发送。Redis 仍会逐条处理命令，但客户端少等了很多次来回。

Pipeline 减少网络往返但不保证业务原子性；MULTI/EXEC 连续执行且没有回滚；WATCH 发现版本变化后需要冲突重试。

如果这些命令彼此独立，只想减少往返，明确使用 transaction=False：

async def load_products_from_cache(client, product_ids: list[int]):
    async with client.pipeline(transaction=False) as pipe:
        for product_id in product_ids:
            pipe.get(product_cache_key(product_id))
        return await pipe.execute()

异步 pipeline 中，入队的 pipe.get() 不需要 await，真正发出命令的是 await pipe.execute()。一次塞太多命令会同时占用客户端输出缓冲、服务端输入缓冲和结果内存，所以大批量任务应分块。块大小要通过数据体积和压测决定，不要只看命令条数。

事务保证连续执行，不提供回滚

transaction=True 会使用 MULTI/EXEC。事务中的命令会连续执行，不会被另一个客户端的命令插到中间。但 Redis 事务没有传统关系数据库那样的回滚：某条命令因为类型错误而失败，其他已经排队的命令仍可能执行。

因此，事务适合把一组已经验证过参数的简单命令作为连续单元提交，不适合把大量复杂业务逻辑塞进去再期待出错时恢复现场。事务里也不要执行可能长时间阻塞的命令，它会把后续请求一起拖住。

WATCH 保护读后再写

扣减库存需要先读当前数量，再决定能不能写。如果两个请求同时读到库存 5，各自扣 4，普通的“GET 后 SET”会产生丢失更新。WATCH 会观察键：从读取到 EXEC 之间只要有人修改它，本次事务就抛出 WatchError，应用重新读取并判断。

from redis.exceptions import WatchError
 
 
async def reserve_stock(client, sku_id: int, order_id: str, quantity: int):
    if quantity <= 0:
        raise ValueError("预留数量必须大于 0")
 
    tag = f"sku:{sku_id}"
    stock_key = f"prod:stock:{{{tag}}}:available"
    order_key 

库存键和订单预留键使用同一哈希标签，让 Cluster 事务可以落到同一槽。冲突重试必须有上限；热点极高时，大量客户端反复 WATCH 会浪费工作，可以考虑把短小逻辑改为 Lua、Redis Function，或者从业务上串行化同一商品的写入。

错误处理要按业务后果分级

别用一个 RedisError 吃掉所有问题

连接错误和超时通常来自短暂网络或服务状态，可能重试或降级；ResponseError 常意味着命令参数不对、键类型冲突或权限不足，继续重试不会把错误代码变正确；序列化失败则应检查数据和版本。把所有异常都捕获后返回 None，会把程序缺陷伪装成缓存未命中。

一个实用的分类可以这样写：

import logging
 
from redis.exceptions import (
    AuthenticationError,
    ConnectionError,
    DataError,
    ResponseError,
    TimeoutError,
)
 
 
logger = logging.getLogger(__name__)
 
 
async def read_optional_cache(client, key: str) -> str | None:
    try:
        return await client.get(key)
    except AuthenticationError:

日志里记录命令类别、依赖名、耗时、异常类型和追踪 ID 即可。不要记录密码、Redis URL、会话号、完整键值或用户数据。键名如果含租户或对象标识，也应按可观测性规范做归一化，避免指标基数爆炸。

同一个故障，不同用途有不同答案

降级不是“捕获异常然后什么也不做”。你要写清楚替代数据源、并发上限、返回语义、告警阈值和恢复后的补偿动作。

用测试把时序问题逼出来

Redis 接入最危险的 bug 往往不在单次成功调用里，而在过期、并发和故障交界处。只测“写入后能读到”远远不够。

单元测试验证业务分支

仓储接口和缓存适配器可以用 stub 或 fake 验证这些分支：

• 缓存命中时不访问数据库；
• 缓存未命中时只回源一次，并按预期 TTL 回填；
• 不存在哨兵使用更短 TTL；
• Redis 读写失败时执行正确降级；
• 反序列化失败不会被当成普通未命中；
• 更新数据库成功后会请求删除缓存。

Fake 能让测试快，但它通常不会完整复现连接池、Lua、事务、Cluster 槽位和真实过期行为。所以关键路径还要有真实 Redis 的集成测试。测试实例必须与开发或生产隔离，每个用例使用唯一前缀；除非实例就是本次测试专用，不要用 FLUSHDB 清场。

集成测试验证真实语义

可以按下面的清单组织：

压测也要包含缓存冷启动、热门键同时过期、连接池耗尽和 Redis 恢复后的重试潮。只在全命中状态下压一个 GET，测出来的是 Redis 的理想路径，不是你的应用会遇到的完整故事。

上线前补齐观测和安全

上线前同时守住异常处理、测试覆盖、可观测性与安全边界，Redis 才算真正接稳。

观测客户端，而不只盯服务端

Redis 服务端 CPU 不高，不代表应用侧没问题。连接池等待、DNS 或网络延迟、客户端超时、重试放大都可能发生在服务端指标之外。应用至少应观察：

• 各命令类别的调用量、耗时分布和错误数；
• 连接创建失败、连接池等待与池耗尽；
• 自动重试次数，以及重试后的最终结果；
• 缓存命中、未命中、回源和回填失败；
• 会话读取失败、幂等冲突、WATCH 冲突与重试耗尽；
• 数据库回源并发和 Redis 故障期间的降级比例。

redis-py 可以接入 OpenTelemetry。即使暂时不启用完整客户端遥测，也应在自己的缓存仓储层打统一 span 和指标。指标标签使用规范化命令名、用途和状态，别把原始键名或用户 ID 作为标签。

慢命令分析也不能只看平均值。大范围 LRANGE、一次返回太多成员的 ZRANGE、巨型 Hash 的 HGETALL、生产环境的 KEYS，都可能让单次请求占用过久。应用层应限制输入范围，并让返回数据量有上限。

Redis 不应该直接面对不可信网络

应用连接 Redis 时，基础安全线包括：

• Redis 只在可信网络内可达，不把端口直接暴露到公网；
• 跨主机通信使用 TLS，并验证服务端证书；
• 使用 ACL 命名用户，按服务授予需要的命令类别和键前缀；
• 密钥通过密钥管理系统或运行环境注入，不写源码、不进镜像、不打日志；
• 管理命令与业务账号分离，应用账号不应拥有 CONFIG、FLUSHALL 等权限；
• 数据按敏感级别决定是否进入 Redis，并结合持久化、备份和淘汰策略评估暴露面。

ACL 最小权限要跟功能一起演进。新代码增加 Stream 或脚本后，如果权限不足，应在预发布环境明确失败，而不是到生产才把 NOPERM 当成普通缓存故障吞掉。

一份可执行的上线检查

• 客户端在应用启动时创建、关停时 aclose()，请求路径不新建连接；
• 连接和命令超时小于接口总预算，重试次数有上限并带退避抖动；
• 每种用途明确事实源、降级方式和不确定写入的处理；
• 缓存有版本化键名、合理 TTL、失效策略和雪崩保护；
• 临时键都有寿命，永久键经过主动评审；
• pipeline 分块，事务范围短，WATCH 冲突有上限；
• 会话 ID 可轮换，Cookie 安全属性齐全，Redis 故障不绕过认证；
• 集成测试覆盖过期、并发、断连、超时、权限和错误键类型；
• 指标能区分命中、未命中、依赖故障、降级和重试；
• Redis 使用私网、TLS、ACL 与密钥托管，日志不记录敏感值。

检查你是否真的掌握了边界