Python缓存读写案例有哪些？

wen python案例 2026-06-07 19:52:46 1

Python缓存读写案例有哪些？用这5个实战方案提升性能（附代码）

目录导读

为什么Python缓存如此重要？
基础缓存：字典与functools.lru_cache
文件缓存：使用pickle持久化数据
分布式缓存：Redis读写实战
内存缓存：cachetools与LRU淘汰策略
数据库缓存：SQLite与sqlite3结合
常见问题与最佳实践

为什么Python缓存如此重要？

核心观点：缓存通过存储昂贵计算的结果，减少重复I/O或CPU消耗，是性能优化的关键。

实际案例：某数据报表系统每天需查询数据库生成50万条汇总，加入Redis缓存后，响应时间从3秒降至50毫秒，数据库负载下降90%。

问答环节：

Q: 缓存会不会导致数据不一致？
A: 会，需要为缓存设置合理的过期时间（TTL）或主动失效机制（例如数据更新时清除缓存）。

基础缓存：字典与`functools.lru_cache`

1 手动字典缓存

cache = {}
def get_user(user_id):
    if user_id in cache:
        return cache[user_id]
    result = fetch_from_db(user_id)  # 假设耗时操作
    cache[user_id] = result
    return result

优点：简单直接，适合小规模场景。
缺点：无内存上限控制，可能内存泄漏。

2 `lru_cache`装饰器

from functools import lru_cache
@lru_cache(maxsize=128)  # 缓存最近128次调用
def compute_fib(n):
    if n < 2:
        return n
    return compute_fib(n-1) + compute_fib(n-2)

工作原理：基于LRU（最近最少使用）淘汰策略，Python内部用字典+双向链表实现，性能极高。

问答环节：

Q: lru_cache能用于类方法吗？
A: 可以，但注意它缓存的是(self, args)组合，如果self对象不同，缓存不共享，建议将方法设计为纯函数，或使用实例级缓存。

文件缓存：使用`pickle`持久化数据

适合场景：计算结果需要跨进程或重启后复用，例如机器学习特征工程预计算结果。

import pickle
import os
def cached_computation(key):
    cache_file = f'cache/{key}.pkl'
    if os.path.exists(cache_file):
        with open(cache_file, 'rb') as f:
            return pickle.load(f)
    result = time_consuming_task(key)  # 耗时操作
    os.makedirs('cache', exist_ok=True)
    with open(cache_file, 'wb') as f:
        pickle.dump(result, f)
    return result

优化点：可以配合hashlib将参数哈希为文件名，避免文件名过长。

安全提示：不要对不可信数据使用pickle.load，否则可能执行恶意代码，推荐改用json（但仅支持基本类型）或dill库（支持更多类型）。

分布式缓存：Redis读写实战

1 基础连接与读写

import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def get_product_price(product_id):
    cache_key = f'product:price:{product_id}'
    cached = r.get(cache_key)
    if cached is not None:
        return float(cached)
    price = query_database_for_price(product_id)  # 假设耗时
    r.setex(cache_key, 300, price)  # 设置300秒过期
    return price

2 批量缓存与管道操作

# 批量写入时使用pipeline提升性能（减少网络往返）
pipe = r.pipeline()
for pid in product_ids:
    pipe.get(f'product:price:{pid}')
cached_prices = pipe.execute()

问答环节：

Q: Redis缓存穿透如何解决？
A: 当查询不存在的数据时，可以缓存空值（如-1）并设置短TTL，或使用布隆过滤器（Bloom Filter）提前过滤非法查询。

内存缓存：`cachetools`与LRU淘汰策略

cachetools提供了比lru_cache更灵活的内存缓存，支持TTL、多线程安全。

from cachetools import TTLCache
cache = TTLCache(maxsize=1000, ttl=600)  # 最大1000个缓存，600秒过期
@cache  # 直接装饰函数
def get_data(url):
    return fetch_from_api(url)

高级用法：自定义淘汰策略（如LFU最不经常使用）：

from cachetools import LFUCache
cache = LFUCache(maxsize=500)  # 优先删除访问次数最少的

对比：

lru_cache：无需安装，适合单一函数
cachetools：支持多缓存实例、TTL、线程安全，适合复杂业务

数据库缓存：SQLite与`sqlite3`结合

适合中小型项目,无需额外服务，将计算结果存为SQLite表。

import sqlite3
import hashlib
def cached_sqlite(key):
    conn = sqlite3.connect('cache.db')
    cursor = conn.cursor()
    cursor.execute('''CREATE TABLE IF NOT EXISTS cache 
                      (key_hash TEXT PRIMARY KEY, value TEXT, 
                       expire_time INTEGER)''')
    key_hash = hashlib.sha256(key.encode()).hexdigest()
    cursor.execute('SELECT value, expire_time FROM cache WHERE key_hash=?', 
                   (key_hash,))
    row = cursor.fetchone()
    if row and row[1] > time.time():
        return row[0]
    result = compute(key)
    cursor.execute('REPLACE INTO cache VALUES (?,?,?)',
                   (key_hash, result, time.time()+3600))
    conn.commit()
    return result

优势：无需安装Redis，且支持SQL查询（如清理过期缓存）。

常见问题与最佳实践

1 缓存雪崩、击穿、穿透

通过以下表格清晰对比：

现象	定义	解决方案
穿透	查询不存在的数据，导致缓存无用	缓存空值、布隆过滤器
击穿	缓存过期瞬间大量查询击穿DB	互斥锁、热点数据永不过期
雪崩	缓存大面积同时失效	随机TTL、多级缓存

2 缓存更新策略

失效模式：读时检测TTL，过期则重新拉取
主动刷新：写数据库时同步更新缓存（双写一致性）
异步刷新：通过消息队列（如RabbitMQ）通知缓存更新

3 性能监控

使用cache_info()方法查看缓存命中率：

@lru_cache
def foo(): ...
print(foo.cache_info())  # 输出命中次数、未命中次数等

问答环节：

Q: 如何选择缓存库？
A: 单机函数级用lru_cache；需要多实例共享用cachetools；跨进程用Redis；大数据量用Memcached或Redis集群；持久化场景用文件/数据库缓存。

Python缓存从单线程字典到分布式Redis,各有适用场景，关键决策点为：数据一致性要求、存储规模、并发阈值，建议先用lru_cache快速验证，再根据性能瓶颈升级方案。

标签： LRU缓存

本文地址： https://dfhcn.com/post/1727.html

文章来源： wen

Python缓存读写案例有哪些？