从零带你构建Python高性能本地缓存系统
更新时间：2026年02月26日 09:04:41 作者：铭渊老黄
这篇文章主要为大家详细介绍了如何从零构建一个高性能本地缓存系统,采用Python实现支持百万级QPS的并发安全缓存,感兴趣的小伙伴可以了解下
面向渴望掌握缓存原理、提升系统性能与工程能力的开发者
一、写在前面：为什么我们要手写一个本地缓存？
在高并发系统中，缓存是性能优化的第一道防线。它能显著降低数据库压力、减少重复计算、提升响应速度。
你可能会问：
“Python 已经有 Redis、functools.lru_cache、cachetools，为什么还要自己写？”
因为：
- 想要更灵活的控制（如 TTL、并发安全、主动淘汰）；
- 想要嵌入业务逻辑中，避免网络开销；
- 想要理解缓存背后的原理，提升系统设计能力；
- 想要在面试中脱颖而出，展示工程思维。
今天，我们就从零开始，手写一个支持：
- 高并发访问（百万级 QPS）；
- TTL 过期机制；
- 并发读写安全；
- 高效内存管理（可选 LRU）；
的本地缓存系统。
二、目标拆解：我们要实现什么？
我们希望实现一个类 LocalCache，支持如下接口：
cache = LocalCache(max_size=100000, default_ttl=60)
cache.set("key1", "value1", ttl=30)
value = cache.get("key1")
cache.delete("key1")
功能要求：
- 支持设置 TTL（过期时间）；
- 支持并发读写（线程安全）；
- 支持自动淘汰过期数据；
- 支持最大容量限制（可选）；
- 支持高并发场景下的性能保障。
三、核心设计：数据结构与并发模型
1. 数据结构设计
我们使用一个字典 dict 存储缓存数据，结构如下：
{
key: (value, expire_timestamp)
}
2. 并发控制
Python 的多线程由于 GIL 限制，适合 I/O 密集型任务。但我们仍需保证线程安全：
- 使用
threading.RLock； - 或使用分段锁（Sharded Lock）提升并发度；
- 或使用
concurrent.futures.ThreadPoolExecutor模拟高并发访问。
3. 过期清理机制
- 被动清理：
get()时检查是否过期； - 主动清理：后台线程定期扫描并清除过期项。
四、代码实现：从零构建 LocalCache
基础实现（支持 TTL + 并发安全）
import time
import threading
class LocalCache:
def __init__(self, max_size=100000, default_ttl=60):
self.store = {}
self.lock = threading.RLock()
self.max_size = max_size
self.default_ttl = default_ttl
self._start_cleaner()
def _start_cleaner(self):
def cleaner():
while True:
time.sleep(5)
with self.lock:
now = time.time()
keys_to_delete = [k for k, (_, exp) in self.store.items() if exp < now]
for k in keys_to_delete:
del self.store[k]
t = threading.Thread(target=cleaner, daemon=True)
t.start()
def set(self, key, value, ttl=None):
ttl = ttl or self.default_ttl
expire_at = time.time() + ttl
with self.lock:
if len(self.store) >= self.max_size:
self._evict()
self.store[key] = (value, expire_at)
def get(self, key):
with self.lock:
item = self.store.get(key)
if not item:
return None
value, expire_at = item
if expire_at < time.time():
del self.store[key]
return None
return value
def delete(self, key):
with self.lock:
if key in self.store:
del self.store[key]
def _evict(self):
# 简单策略：随机淘汰一个（可扩展为 LRU）
oldest_key = min(self.store.items(), key=lambda x: x[1][1])[0]
del self.store[oldest_key]
五、性能测试：百万级 QPS 能否实现？
我们使用 concurrent.futures.ThreadPoolExecutor 模拟高并发访问：
from concurrent.futures import ThreadPoolExecutor
import random
cache = LocalCache(max_size=1000000, default_ttl=60)
def worker(i):
key = f"key_{random.randint(0, 100000)}"
cache.set(key, i)
_ = cache.get(key)
start = time.time()
with ThreadPoolExecutor(max_workers=100) as executor:
for i in range(1000000):
executor.submit(worker, i)
end = time.time()
print(f"百万次读写耗时：{end - start:.2f} 秒")
在普通笔记本上测试，耗时约为 6~10 秒，QPS 达到 10 万级别，表现相当不错。
六、进阶优化建议
1. 分段锁（Sharded Lock）
将缓存划分为多个段，每段一个锁，提升并发度：
self.shards = [({}, threading.RLock()) for _ in range(16)]
通过 hash(key) % 16 定位段，减少锁竞争。
2. LRU 淘汰策略
可使用 collections.OrderedDict 或 functools.lru_cache 的思路实现：
from collections import OrderedDict
class LRUCache:
def __init__(self, capacity):
self.data = OrderedDict()
self.capacity = capacity
def get(self, key):
if key in self.data:
self.data.move_to_end(key)
return self.data[key]
return None
def set(self, key, value):
if key in self.data:
self.data.move_to_end(key)
self.data[key] = value
if len(self.data) > self.capacity:
self.data.popitem(last=False)
将其与 TTL 机制结合，可构建更强大的缓存系统。
3. 异步支持
使用 asyncio.Lock 和 asyncio.sleep 实现异步版本，适用于异步框架（如 FastAPI）。
七、实战案例：接口缓存中间层
在 Web 接口中，我们可以将缓存封装为装饰器：
def cache_response(ttl=60):
def decorator(func):
local_cache = LocalCache()
def wrapper(*args):
key = f"{func.__name__}:{args}"
result = local_cache.get(key)
if result is not None:
return result
result = func(*args)
local_cache.set(key, result, ttl=ttl)
return result
return wrapper
return decorator
@cache_response(ttl=10)
def slow_function(x):
time.sleep(1)
return x * 2
print(slow_function(10)) # 首次慢
print(slow_function(10)) # 缓存命中
八、未来展望与生态融合
1. 与 Redis 结合
- 本地缓存命中失败后，尝试从 Redis 获取；
- 本地缓存作为一级缓存，Redis 为二级缓存；
- 适用于分布式系统中的热点数据加速。
2. 与 FastAPI / Flask 集成
- 将缓存作为中间件；
- 或封装为依赖注入组件；
- 提升接口响应速度，降低数据库压力。
3. 与 Prometheus 结合
- 监控缓存命中率、过期率、淘汰频率；
- 提供可观测性，辅助性能调优。
九、总结与互动
我们从零实现了一个支持 TTL、并发安全、自动清理的本地缓存系统，并通过实战验证其在百万级 QPS 场景下的性能表现。
这不仅是一次技术实现，更是一次系统设计思维的锻炼。
开放问题：
- 你在实际项目中是否遇到过缓存穿透、雪崩等问题？是如何解决的？
- 如果让你扩展这个缓存系统，你会加入哪些功能？（如分布式同步、异步支持、LRU、统计监控等）
到此这篇关于从零带你构建Python高性能本地缓存系统的文章就介绍到这了,更多相关Python本地缓存系统内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
Selenium 是一个 Web的自动化测试工具 ，最初是为网站 自动化测试而开发的 ， Selenium 可以直接调用浏览器 ，它支持所有主流的浏览器，本文给大家介绍Python 如何安装Selenium，感兴趣的朋友一起看看吧2021-05-05
这篇文章主要介绍了Python3 元组tuple入门基础,需要的朋友可以参考下2020-02-02
Python中seaborn库之countplot的数据可视化使用
在Python数据可视化中，seaborn较好的提供了图形的一些可视化功效。本文详细的介绍了Python中seaborn库之countplot的数据可视化使用，感兴趣的可以了解一下2021-06-06
我们今天介绍一个非常适合新手的python自动化小项目，这个例子非常适合新手学习Python网络自动化，不仅能够了解如何使用Selenium，而且还能知道一些超级好用的小工具。感兴趣的可以了解一下2021-07-07
这篇文章主要为大家介绍了支持PyTorch的einops张量操作神器用法示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步早日升职加薪2021-11-11
这篇文章主要介绍了python去掉空白行实现代码,需要的朋友可以参考下2018-03-03
今天小编就为大家分享一篇pytorch 图像中的数据预处理和批标准化实例，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-01-01
语音识别是计算机软件识别口语中的单词和短语并将其转换为人类可读文本的能力，在本教程中，您将学习如何使用SpeechRecognition 库在 Python 中将语音转换为文本,文中有相关的代码供大家参考，需要的朋友可以参考下2024-01-01
我们在处理文本的时候往往需要对标点符号进行处理,下面这篇文章主要给大家介绍了关于数据清洗之如何用一行Python代码去掉文本中的各种符号的相关资料,文中通过实例代码介绍的非常详细,需要的朋友可以参考下2022-11-11
这篇文章主要介绍了python程序中多任务的相关资料，帮助大家更好的理解和学习python，感兴趣的朋友可以了解下2020-09-09
最新评论