基于《第六章 · AI掘金头条-缓存和模型调用》讲义结合toutiao_backend项目实战代码整理。一、为什么头条类应用必须上缓存想象一下首页同时有 1000 人在刷新闻分类和列表。如果每次请求都直打 MySQL用户 / 前端 → FastAPI业务逻辑中心→ 数据库数据库会变成瓶颈连接池打满、查询排队、接口变慢。缓存就是夹在中间的一层「高速临时数据存储」把已经查过、短期内不会大变的结果放进内存下次直接读缓存而不是再走一遍昂贵的数据库查询。核心收益很直接提升性能与体验减轻数据库负载降低延迟节省资源与成本一句话数据库负责「存得住」缓存负责「取得快」。二、项目里的缓存读写流程讲义里的标准流程也正是本项目采用的路径前端请求 → 服务器查缓存 ├─ 命中直接返回 └─ 未命中查数据库 → 写入缓存 → 返回前端对应到新闻模块大致分三层层级职责项目位置配置层Redis 连接、通用 get/setconfig/cache_conf.py缓存层业务 Key 设计、TTLcache/news_cache.py业务层Cache-Aside 读写编排crud/news_cache.py路由层对外 APIrouters/news.py三、接入 Redis四步落地讲义把 Redis 集成拆成四步项目也按这个节奏实现安装并启动 Redis 服务端默认端口6379安装并配置客户端封装缓存操作设计缓存策略1. 配置异步 Redis 客户端项目使用redis.asyncio与 FastAPI 异步风格一致# config/cache_conf.pyimportredis.asyncioasredis redis_clientredis.Redis(hostlocalhost,port6379,db0,decode_responsesTrue# 字节自动解码为字符串)要点host/port连到哪里db0~15 号逻辑库decode_responsesTrue业务层直接拿字符串少一层编解码麻烦2. 封装「存 / 取」通用方法缓存操作本质就是围绕 Redis 做存、取、删、判断、过期。项目封装了最常用的读字符串、读 JSON、写带过期# 读 JSON列表 / 字典asyncdefget_json_cache(key:str):dataawaitredis_client.get(key)returnjson.loads(data)ifdataelseNone# 写缓存dict/list 先序列化再用 setex 带过期asyncdefset_cache(key:str,value:Any,expire:int3600):ifisinstance(value,(dict,list)):valuejson.dumps(value,ensure_asciiFalse)awaitredis_client.setex(key,expire,value)对应 Redis 原语方法作用setex(key, expire, value)写入并设置过期秒数get(key)读取不存在返回Nonedelete(key)删除指定键异常被吞掉并返回None/False避免 Redis 抖动把整个接口拖垮——缓存失败时业务仍可回落到数据库。四、旁路缓存Cache-Aside最常用的策略讲义强调的「旁路策略」是应用主动管缓存读先查缓存 → 有则返回 → 没有查库 → 回填缓存写先改数据库 → 再更新或删除缓存以新闻分类为例crud/news_cache.pyasyncdefget_categories(db:AsyncSession,skip:int0,limit:int100):# 1. 先查缓存cachedawaitget_cached_categories()ifcached:returncached# 2. 未命中查数据库stmtselect(Category).offset(skip).limit(limit)resultawaitdb.execute(stmt)categoriesresult.scalars().all()# 3. ORM → 可 JSON 化结构再回填缓存ifcategories:categoriesjsonable_encoder(categories)awaitset_cache_categories(categories)returncategories列表、详情、相关新闻同一套路命中缓存 → 必要时把 dict 还原成News(**item)再给上层用未命中 → SQLAlchemy 查库用 Pydanticmodel_dump(by_aliasFalse, modejson)转成「后端友好」字段名再写入 Redis注意by_aliasFalse很关键——缓存是给后端复用的应用 Python/数据库字段风格如publish_time而不是前端别名publishTime。五、缓存 Key 与 TTL策略比「会不会用 Redis」更重要1. Key 设计要稳定、可拼装# cache/news_cache.pyCATEGORIES_KEYnews:categoriesNEWS_LIST_PREFIXnews_list:# news_list:{category}:{page}:{size}NEWS_DETAIL_PREFIXnews:detail:# news:detail:{id}RELATED_NEWS_PREFIXnews:related:# news:related:{news_id}:{category_id}列表 Key 把「分类 页码 每页条数」拼进去保证不同查询条件互不污染keyfnews_list:{category_part}:{page}:{size}2. TTL 按「数据稳定程度」分层避免雪崩讲义原则数据越稳定缓存越久变化越快缓存越短。不同 Key 不要同一时刻集体过期。类型建议 TTL项目默认分类 / 配置7200s2hset_cache_categories(..., 7200)列表600~1800s列表写入默认 1800s详情较短热数据会变详情默认 300s相关新闻1800scache_related_news(..., 1800)验证码类120s讲义示例本模块未实现这也是为什么项目注释写着「避免所有 key 同时过期引起缓存雪崩」。六、接口层怎么用上缓存路由本身不感知 Redis只依赖带缓存的 CRUD# routers/news.pyrouter.get(/categories)asyncdefget_categories(...):categoriesawaitnews_cache.get_categories(db,skip,limit)return{code:200,message:获取新闻分类成功,data:categories}router.get(/list)asyncdefget_news_list(...):news_listawaitnews_cache.get_news_list(db,category_id,offset,page_size)...对比无缓存的crud/news.py与带缓存的crud/news_cache.py你会看到业务 SQL 几乎不变变的是「外面包了一层 Cache-Aside」。这是工程上很舒服的演进方式先跑通 CRUD再无痛加缓存。详情接口还叠了浏览量 1 和相关新闻查详情可缓存→ 浏览量 1写库→ 相关新闻可缓存→ 组装响应相关新闻按「同分类、排除自己、按浏览量与发布时间排序」取 Top N同样走缓存避免详情页二次打库。七、调用通义千问Qwen大模型讲义后半段转向 AI 能力前端可直接对接阿里云百炼上的千问系列。使用步骤讲义流程登录阿里云百炼模型广场选择模型如通义千问 3-Max打开 API 参考拿到HTTP 请求地址与API Key在前端配置并调用前端配置位置前端项目 → src/config/api.js替换三样东西即可大模型 HTTP 请求地址API Key模型名称这样头条产品里就能接上「摘要、问答、推荐文案」等 AI 能力而后端继续用 Redis 扛住新闻读写的高频流量——缓存保稳定模型添智能两条线各司其职。安全提示生产环境不要把 API Key 裸写进前端仓库更稳妥的做法是走后端代理转发Key 放服务端环境变量。课程演示阶段可先按讲义在前端配置上线前务必收口。八、一张图串起本章安装 Redis → 配置异步客户端 → 封装 get/setex ↓ 设计 Key 分层 TTL ↓ Cache-Aside先缓存后数据库 ↓ 分类 / 列表 / 详情 / 相关新闻全覆盖 ↓ 前端对接百炼 Qwen API九、小结缓存是高速临时层用来换性能、降延迟、减库压。Redis内存 Key-Value很适合做 FastAPI 应用层缓存。集成四步服务端 → 客户端 → 封装操作 → 设计策略。项目落地的是Cache-Aside读穿回填、写侧维护一致性。Key 要可区分TTL 要分层这是防雪崩的关键。Qwen经阿里云百炼拿地址与 Key在前端api.js配置后即可调用。如果你正在跟做「AI掘金头条」建议按这个顺序自测redis-cli ping→ 打/api/news/categories两次看第二次是否不再打库日志 → 再换列表/详情验证不同 Key 与过期时间。缓存跑通后再接千问整章目标就齐了哦。