Python脚本如何统计本地缓存存储数量

wen python案例 2026-06-26 242

本文目录导读：

使用 functools.lru_cache（函数级缓存）
使用 cachetools 库（支持多种缓存策略）
使用 redis 作为本地缓存
使用 diskcache（文件系统缓存）
自定义字典缓存统计
监控缓存使用情况
选择建议

在Python中统计本地缓存存储数量,具体取决于你使用的缓存方式，以下是几种常见场景的实现方法：

使用 `functools.lru_cache`（函数级缓存）

from functools import lru_cache
@lru_cache(maxsize=128)
def expensive_function(n):
    return n * n
# 调用函数触发缓存
for i in range(10):
    expensive_function(i)
# 获取缓存信息
cache_info = expensive_function.cache_info()
print(f"缓存命中次数: {cache_info.hits}")
print(f"缓存未命中次数: {cache_info.misses}")
print(f"当前缓存大小: {cache_info.currsize}")
print(f"最大缓存容量: {cache_info.maxsize}")

使用 `cachetools` 库（支持多种缓存策略）

from cachetools import LRUCache, TTLCache
# 创建LRU缓存
cache = LRUCache(maxsize=100)
# 添加数据
cache['key1'] = 'value1'
cache['key2'] = 'value2'
# 统计缓存数量
cache_count = len(cache)
print(f"缓存中项目数量: {cache_count}")
# 或使用currsize属性
print(f"当前缓存大小: {cache.currsize}")
# 带过期时间的缓存
ttl_cache = TTLCache(maxsize=100, ttl=300)
ttl_cache['key1'] = 'value1'
print(f"TTL缓存数量: {len(ttl_cache)}")

使用 `redis` 作为本地缓存

import redis
# 连接本地Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 设置缓存
r.set('user:1', 'data1')
r.set('user:2', 'data2')
# 统计键数量（所有缓存）
cache_count = r.dbsize()
print(f"Redis缓存总数: {cache_count}")
# 统计特定模式的键数量
keys = r.keys('user:*')
print(f"用户缓存数量: {len(keys)}")
# 使用SCAN命令（推荐，不会阻塞）
cursor = 0
count = 0
while cursor != 0 or count == 0:
    cursor, data = r.scan(cursor=cursor, match='user:*', count=1000)
    count += len(data)
print(f"用户缓存数量(scan方式): {count}")

使用 `diskcache`（文件系统缓存）

import diskcache as dc
# 创建文件缓存
cache = dc.Cache('/tmp/my_cache')
# 添加缓存
cache['key1'] = 'value1'
cache['key2'] = 'value2'
# 统计缓存数量
cache_count = len(cache)
print(f"磁盘缓存数量: {cache_count}")
# 或使用volume方法获取详情
stats = cache.volume()
print(f"缓存统计: {stats}")

自定义字典缓存统计

class CacheStats(dict):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.hits = 0
        self.misses = 0
    def get(self, key, default=None):
        if key in self:
            self.hits += 1
        else:
            self.misses += 1
        return super().get(key, default)
    def stats(self):
        return {
            'size': len(self),
            'hits': self.hits,
            'misses': self.misses,
            'hit_rate': self.hits / (self.hits + self.misses) if (self.hits + self.misses) > 0 else 0
        }
# 使用示例
cache = CacheStats()
cache['user1'] = 'data1'
cache['user2'] = 'data2'
cache.get('user1')  # 命中
cache.get('user3')  # 未命中
print(f"缓存统计: {cache.stats()}")

监控缓存使用情况

import time
from collections import defaultdict
class CacheMonitor:
    def __init__(self):
        self.caches = defaultdict(dict)
    def register_cache(self, name, cache_obj):
        self.caches[name] = {
            'cache': cache_obj,
            'last_check': time.time()
        }
    def monitor_all(self):
        results = {}
        for name, info in self.caches.items():
            cache = info['cache']
            if hasattr(cache, 'currsize'):
                size = cache.currsize
            elif hasattr(cache, '__len__'):
                size = len(cache)
            else:
                size = 'N/A'
            results[name] = {
                'size': size,
                'time_since_check': time.time() - info['last_check']
            }
            info['last_check'] = time.time()
        return results
# 使用示例
monitor = CacheMonitor()
monitor.register_cache('lru_cache', expensive_function)
monitor.register_cache('dict_cache', {'a': 1, 'b': 2})
print(monitor.monitor_all())