超级缓存：Cashews 如何在 Python 异步应用中实现“10x 加速

🥔 异步缓存框架，简洁API打造快速可靠的应用 🥔

📝 引言

在现代应用中，缓存不是一种选择，而是一种必需品。但传统的缓存工具往往功能单一，面对缓存雪崩、缓存穿透、缓存击穿等经典问题，开发者需要自己实现各种复杂的保护机制。

今天介绍的开源框架Cashews（土豆），以简洁的装饰器API和强大的异步能力，让这些高级缓存策略变得触手可及。

📊 官方数据显示：在启用客户端缓存的情况下，Cashews比另一个流行框架aiocache快2倍以上。

🥔 Cashews是什么？

Cashews是一个基于异步编程的缓存框架，提供简单易用的API，帮助开发者构建快速、可靠和可扩展的应用程序。

💡 有趣的事实：项目名称“Cashews”（腰果）虽为谐音，但其图标却是🥔（土豆），也许正是暗示它像土豆一样朴实却万能的特性。

✨ 核心特性

📊 与其他框架对比

🚀 快速上手

📦 安装

# 基础安装（内存缓存）pip install cashews# Redis支持（分布式缓存）pip install cashews[redis]# DiskCache支持（本地数据库缓存）pip install cashews[diskcache]# 完整功能（布隆过滤器等）pip install cashews[speedup]

💻 基础用法

from cashews import cacheimport asyncio# 配置内存缓存cache.setup("mem://")# 装饰器模式：自动缓存函数结果@cache(ttl="10m", key="user:{user_id}")asyncdefget_user(user_id: int):await asyncio.sleep(1)  # 模拟数据库查询return {"id": user_id, "name": f"User{user_id}"}asyncdefmain():# 第一次调用：执行函数，耗时1秒    user1 = await get_user(1)print("✅ 第一次调用完成 - 耗时1秒")# 第二次调用：从缓存读取，瞬间返回    user2 = await get_user(1)print("⚡ 第二次调用完成 - 瞬间返回")asyncio.run(main())

✨ 通过装饰器轻松实现缓存逻辑，无需侵入业务代码。

🛡️ 解决分布式缓存三大难题

1. Early Cache：解决缓存雪崩

缓存雪崩指大量缓存同时过期，导致请求直接打到数据库。Cashews提供了early策略，在缓存即将过期时异步更新缓存：

from cashews import cachecache.setup("redis://localhost:6379")# 10分钟过期，7分钟时在后台自动更新@cache.early(ttl="10m", early_ttl="7m")asyncdefget_popular_products():returnawait fetch_products_from_db()

🔒 确保缓存始终有效，彻底避免缓存雪崩。

2. Soft Cache：解决缓存击穿

缓存击穿指热点key失效，高并发请求同时查询数据库。Soft策略允许缓存过期后继续使用旧数据，同时异步更新：

@cache.soft(ttl="10m", soft_ttl="1h")asyncdefget_hot_product(product_id: int):"""1小时内即使缓存过期也返回旧值"""returnawait fetch_product(product_id)

💪 确保系统持续可用，即使后端服务暂时不可用。

3. Locked：解决缓存击穿（另一角度）

确保只有一个请求会执行函数，其他并发请求等待结果：

@cache.locked(ttl="10s")asyncdefexpensive_computation():# 同时被100个请求调用，只有第一个实际执行returnawait compute()# 与缓存组合使用@cache(ttl="1h", lock=True)asyncdefget_expensive_data():"""锁保证单次执行，缓存保证后续快速返回"""returnawait compute()

📉 Fail Cache：服务降级利器

微服务架构中，当服务不可用时，使用缓存数据作为降级方案：

@cache.failover(ttl="30m", exceptions=(TimeoutError, ConnectionError))asyncdeffetch_from_microservice(service_id: str):"""调用微服务API，失败时返回缓存的旧数据"""returnawait call_service(service_id)

🚨 只要有一次成功调用，当服务再次异常时，自动返回缓存数据，实现优雅的服务降级。

⏱️ 限流与熔断

速率限制

from cashews import cache, RateLimitError# 每分钟最多10次调用，超限后禁用10分钟@cache.rate_limit(limit=10, period="1m", ttl="10m")asyncdefget(name):returnawait api_call()# 与故障转移组合@cache.failover(ttl="10m", exceptions=(RateLimitError,))@cache.slice_rate_limit(limit=100, period="10m")asyncdefget_next(name):"""限流时返回缓存数据"""returnawait api_call()

熔断器

from cashews import cache, CircuitBreakerOpen# 1分钟内错误率超过10%时熔断5分钟@cache.circuit_breaker(errors_rate=10, period="1m", ttl="5m")asyncdefget(name):returnawait api_call()

🎨 灵活的Key模板

基础模板

@cache(ttl="2h", key="user_info:{user_id}")asyncdefget_info(user_id: str):    ...@cache(ttl="2h", key="user_info:{user.uuid}")asyncdefget_info(user: User):    ...

内置格式化函数

@cache(ttl="2h", key="user_info:{user.name:lower}:{password:hash(sha1)}")asyncdefget_info(user: User, password: str):    ...@cache(ttl="2h", key="user:{token:jwt(client_id)}")asyncdefget_user_by_token(token: str):    ...

💾 多后端存储

基于前缀的路由

from cashews import cache# 根据不同前缀使用不同后端cache.setup("redis://redis/0")                    # 默认Rediscache.setup("mem://?size=500", prefix="session")  # session使用内存cache.setup("disk://?directory=/tmp/cache", prefix="file")  # 文件使用磁盘# 使用方式完全一致await cache.get("accounts")        # 从Redis获取await cache.get("session:1")       # 从内存获取await cache.get("file:resume.pdf") # 从磁盘获取

Redis客户端缓存（10倍性能提升）

# 启用客户端缓存，热点数据缓存在应用本地内存cache.setup("redis://localhost:6379", client_side=True)@cache(ttl="5m")asyncdefget_frequently_accessed_data(key: str):returnawait expensive_operation(key)

🧹 缓存失效策略

标签失效

避免扫描整个数据库，使用SET存储标签关联的键：

from cashews import cachecache.setup("redis://", client_side=True)@cache(ttl="1h", tags=["items", "page:{page}"])asyncdefitems(page=1):    ...# 按标签批量失效await cache.delete_tags("page:1")  # 只失效第1页await cache.delete_tags("items")    # 失效所有items相关的缓存

代码变更自动失效

from dataclasses import dataclass@dataclassclassUser:    name: str    surname: str# 使用数据类，结构变化时__repr__自动变化，缓存失效@cache(ttl="1d")asyncdefget_user(user_id):return User("Dima", "Krykov")

🔍 检测结果来源

from cashews import cachewith cache.detect as detector:    response = await something_that_use_cache()    calls = detector.calls# 输出示例：{"my:key": [{"ttl": 10, "name": "simple", "backend": "redis"}]}

FastAPI中间件示例

@app.middleware("http")asyncdefadd_from_cache_headers(request: Request, call_next):with cache.detect as detector:        response = await call_next(request)if detector.calls:            key = list(detector.calls.keys())[0]            response.headers["X-From-Cache"] = keyreturn response

⚡ FastAPI集成

中间件

from fastapi import FastAPI, Header, Queryfrom fastapi.responses import StreamingResponsefrom cashews import cachefrom cashews.contrib.fastapi import (    CacheDeleteMiddleware,    CacheEtagMiddleware,    CacheRequestControlMiddleware,    cache_control_ttl,)app = FastAPI()app.add_middleware(CacheDeleteMiddleware)app.add_middleware(CacheEtagMiddleware)app.add_middleware(CacheRequestControlMiddleware)cache.setup(os.environ.get("CACHE_URI", "redis://"))@app.get("/")@cache.failover(ttl="1h")@cache(ttl=cache_control_ttl(default="4m"), key="simple:{user_agent:hash}", time_condition="1s")asyncdefsimple(user_agent: str = Header("No")):    ...@app.get("/stream")@cache(ttl="1m", key="stream:{file_path}")asyncdefstream(file_path: str = Query(__file__)):return StreamingResponse(_read_file(file_path))

📊 Prometheus监控

from cashews import cachefrom cashews.contrib.prometheus import create_metrics_middlewaremetrics_middleware = create_metrics_middleware(with_tag=False)cache.setup("redis://", middlewares=(metrics_middleware,))

🎬 结语

Cashews不仅是一个缓存库，更是一套完整的缓存策略解决方案。

它用简洁的装饰器API封装了复杂的缓存最佳实践，让开发者可以轻松应对高并发场景下的各种挑战。

无论你是正在构建新的异步应用，还是希望优化现有系统的性能，Cashews都值得一试。正如其名，这个小小的“土豆”可能会成为你项目中的秘密武器。

📚 资源链接

💬 互动话题

你在项目中遇到过哪些缓存难题？欢迎在评论区分享你的经验！