IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python脚本如何实现本地缓存读写操作

wen python案例 29

本文目录导读：

Python脚本如何实现本地缓存读写操作

  1. 使用字典实现简单内存缓存
  2. 使用 functools.lru_cache (装饰器方式)
  3. 使用 JSON 文件持久化缓存
  4. 使用 pickle 序列化（支持复杂对象）
  5. 使用第三方库 cachetools
  6. 实际应用示例：带自动刷新的缓存
  7. 使用建议

我来介绍几种Python实现本地缓存读写的方法：

使用字典实现简单内存缓存

import time
import threading
class SimpleCache:
    def __init__(self, default_ttl=300):  # 默认过期时间5分钟
        self.cache = {}
        self.ttl = default_ttl
        self.lock = threading.Lock()
    def get(self, key):
        with self.lock:
            if key in self.cache:
                data, expire_time = self.cache[key]
                if time.time() < expire_time:
                    return data
                else:
                    del self.cache[key]
        return None
    def set(self, key, value, ttl=None):
        if ttl is None:
            ttl = self.ttl
        expire_time = time.time() + ttl
        with self.lock:
            self.cache[key] = (value, expire_time)
    def delete(self, key):
        with self.lock:
            if key in self.cache:
                del self.cache[key]
    def clear(self):
        with self.lock:
            self.cache.clear()
# 使用示例
cache = SimpleCache()
cache.set("user_1", {"name": "张三", "age": 25})
data = cache.get("user_1")
print(data)  # {"name": "张三", "age": 25}

使用 functools.lru_cache (装饰器方式)

from functools import lru_cache
import time
@lru_cache(maxsize=100)
def expensive_function(n):
    """模拟耗时计算"""
    time.sleep(2)  # 模拟耗时操作
    return n * n
# 第一次调用，会执行函数
print(expensive_function(5))  # 耗时2秒
# 第二次调用，直接返回缓存结果
print(expensive_function(5))  # 立即返回
# 查看缓存信息
print(expensive_function.cache_info())
# CacheInfo(hits=1, misses=1, maxsize=100, currsize=1)
# 清除缓存
expensive_function.cache_clear()

使用 JSON 文件持久化缓存

import json
import os
import time
from typing import Any, Optional
class FileCache:
    def __init__(self, cache_file="cache.json", default_ttl=3600):
        self.cache_file = cache_file
        self.default_ttl = default_ttl
        self._load_cache()
    def _load_cache(self):
        if os.path.exists(self.cache_file):
            with open(self.cache_file, 'r', encoding='utf-8') as f:
                self.cache = json.load(f)
        else:
            self.cache = {}
    def _save_cache(self):
        with open(self.cache_file, 'w', encoding='utf-8') as f:
            json.dump(self.cache, f, ensure_ascii=False, indent=2)
    def get(self, key: str) -> Optional[Any]:
        if key in self.cache:
            data = self.cache[key]
            expire_time = data.get('expire_time')
            if expire_time is None or time.time() < expire_time:
                return data['value']
            else:
                # 过期删除
                del self.cache[key]
                self._save_cache()
        return None
    def set(self, key: str, value: Any, ttl: Optional[int] = None):
        if ttl is None:
            ttl = self.default_ttl
        self.cache[key] = {
            'value': value,
            'expire_time': time.time() + ttl,
            'create_time': time.time()
        }
        self._save_cache()
    def delete(self, key: str):
        if key in self.cache:
            del self.cache[key]
            self._save_cache()
    def clear(self):
        self.cache = {}
        self._save_cache()
# 使用示例
file_cache = FileCache("my_cache.json")
file_cache.set("config", {"theme": "dark", "language": "zh-CN"})
data = file_cache.get("config")
print(data)  # {"theme": "dark", "language": "zh-CN"}

使用 pickle 序列化（支持复杂对象）

import pickle
import os
import time
class PickleCache:
    def __init__(self, cache_file="cache.pkl"):
        self.cache_file = cache_file
        self._load_cache()
    def _load_cache(self):
        if os.path.exists(self.cache_file):
            with open(self.cache_file, 'rb') as f:
                self.cache = pickle.load(f)
        else:
            self.cache = {}
    def _save_cache(self):
        with open(self.cache_file, 'wb') as f:
            pickle.dump(self.cache, f)
    def get(self, key):
        return self.cache.get(key)
    def set(self, key, value):
        self.cache[key] = value
        self._save_cache()
    def delete(self, key):
        if key in self.cache:
            del self.cache[key]
            self._save_cache()
# 使用示例（支持复杂对象）
class User:
    def __init__(self, name, age):
        self.name = name
        self.age = age
    def __repr__(self):
        return f"User({self.name}, {self.age})"
pickle_cache = PickleCache()
user = User("李四", 30)
pickle_cache.set("user_1", user)
retrieved = pickle_cache.get("user_1")
print(retrieved)  # User(李四, 30)

使用第三方库 cachetools

首先安装：

pip install cachetools
from cachetools import TTLCache, cached
import time
# 创建TTL缓存，最大100个条目，过期时间5分钟
cache = TTLCache(maxsize=100, ttl=300)
@cached(cache)
def get_user_info(user_id):
    """模拟获取用户信息"""
    time.sleep(1)  # 模拟数据库查询
    return {"id": user_id, "name": f"User_{user_id}"}
# 使用
result1 = get_user_info(1)  # 耗时1秒
result2 = get_user_info(1)  # 立即返回缓存
print(result1, result2)
# 缓存统计
print(f"缓存大小: {len(cache)}")
print(f"命中率: {cache.currsize}/{cache.maxsize}")

实际应用示例：带自动刷新的缓存

import time
import threading
from functools import wraps
class AutoRefreshCache:
    """自动刷新缓存装饰器"""
    def __init__(self, ttl=60, maxsize=100):
        self.ttl = ttl
        self.maxsize = maxsize
        self.cache = {}
        self.lock = threading.Lock()
    def __call__(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = str(args) + str(kwargs)
            with self.lock:
                if key in self.cache:
                    data, timestamp = self.cache[key]
                    if time.time() - timestamp < self.ttl:
                        return data
                    # 异步刷新缓存
                    def refresh():
                        try:
                            new_data = func(*args, **kwargs)
                            with self.lock:
                                self.cache[key] = (new_data, time.time())
                        except Exception as e:
                            print(f"缓存刷新失败: {e}")
                    # 启动刷新线程
                    threading.Thread(target=refresh, daemon=True).start()
                    return data  # 返回旧数据
                # 首次调用
                data = func(*args, **kwargs)
                self.cache[key] = (data, time.time())
                # 限制缓存大小
                if len(self.cache) > self.maxsize:
                    oldest = min(self.cache.items(), key=lambda x: x[1][1])
                    del self.cache[oldest[0]]
                return data
        return wrapper
# 使用示例
@AutoRefreshCache(ttl=30, maxsize=10)
def get_api_data(url):
    """模拟API调用"""
    print(f"正在请求: {url}")
    return {"data": f"Response from {url}"}
# 测试
result1 = get_api_data("http://api.example.com")
print(result1)
# 30秒内再次调用，使用缓存
result2 = get_api_data("http://api.example.com")
print(result2)

使用建议

  1. 简单场景：使用字典或 functools.lru_cache
  2. 需要持久化：使用文件缓存（JSON/Pickle）
  3. 需要过期时间：使用 cachetools.TTLCache
  4. 高性能要求：考虑使用 Redis 等外部缓存
  5. 多进程环境：使用文件或数据库缓存

选择哪种方式取决于你的具体需求，包括数据大小、持久性要求、访问频率等因素。

上一篇Python脚本如何设置缓存过期失效时间

下一篇Python脚本如何保证协程任务幂等性

相关文章

抱歉，评论功能暂时关闭!