從零帶你構(gòu)建Python高性能本地緩存系統(tǒng)
更新時間：2026年02月26日 09:04:41 作者：銘淵老黃
這篇文章主要為大家詳細介紹了如何從零構(gòu)建一個高性能本地緩存系統(tǒng),采用Python實現(xiàn)支持百萬級QPS的并發(fā)安全緩存,感興趣的小伙伴可以了解下
面向渴望掌握緩存原理、提升系統(tǒng)性能與工程能力的開發(fā)者
一、寫在前面：為什么我們要手寫一個本地緩存？
在高并發(fā)系統(tǒng)中，緩存是性能優(yōu)化的第一道防線。它能顯著降低數(shù)據(jù)庫壓力、減少重復計算、提升響應速度。
你可能會問：
“Python 已經(jīng)有 Redis、functools.lru_cache、cachetools，為什么還要自己寫？”
因為：
- 想要更靈活的控制（如 TTL、并發(fā)安全、主動淘汰）；
- 想要嵌入業(yè)務邏輯中，避免網(wǎng)絡(luò)開銷；
- 想要理解緩存背后的原理，提升系統(tǒng)設(shè)計能力；
- 想要在面試中脫穎而出，展示工程思維。
今天，我們就從零開始，手寫一個支持：
- 高并發(fā)訪問（百萬級 QPS）；
- TTL 過期機制；
- 并發(fā)讀寫安全；
- 高效內(nèi)存管理（可選 LRU）；
的本地緩存系統(tǒng)。
二、目標拆解：我們要實現(xiàn)什么？
我們希望實現(xiàn)一個類 LocalCache，支持如下接口：
cache = LocalCache(max_size=100000, default_ttl=60)
cache.set("key1", "value1", ttl=30)
value = cache.get("key1")
cache.delete("key1")
功能要求：
- 支持設(shè)置 TTL（過期時間）；
- 支持并發(fā)讀寫（線程安全）；
- 支持自動淘汰過期數(shù)據(jù)；
- 支持最大容量限制（可選）；
- 支持高并發(fā)場景下的性能保障。
三、核心設(shè)計：數(shù)據(jù)結(jié)構(gòu)與并發(fā)模型
1. 數(shù)據(jù)結(jié)構(gòu)設(shè)計
我們使用一個字典 dict 存儲緩存數(shù)據(jù)，結(jié)構(gòu)如下：
{
key: (value, expire_timestamp)
}
2. 并發(fā)控制
Python 的多線程由于 GIL 限制，適合 I/O 密集型任務。但我們?nèi)孕璞ＷC線程安全：
- 使用
threading.RLock； - 或使用分段鎖（Sharded Lock）提升并發(fā)度；
- 或使用
concurrent.futures.ThreadPoolExecutor模擬高并發(fā)訪問。
3. 過期清理機制
- 被動清理：
get()時檢查是否過期； - 主動清理：后臺線程定期掃描并清除過期項。
四、代碼實現(xiàn)：從零構(gòu)建 LocalCache
基礎(chǔ)實現(xiàn)（支持 TTL + 并發(fā)安全）
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 能否實現(xiàn)？
我們使用 concurrent.futures.ThreadPoolExecutor 模擬高并發(fā)訪問：
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 萬級別，表現(xiàn)相當不錯。
六、進階優(yōu)化建議
1. 分段鎖（Sharded Lock）
將緩存劃分為多個段，每段一個鎖，提升并發(fā)度：
self.shards = [({}, threading.RLock()) for _ in range(16)]
通過 hash(key) % 16 定位段，減少鎖競爭。
2. LRU 淘汰策略
可使用 collections.OrderedDict 或 functools.lru_cache 的思路實現(xiàn)：
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 機制結(jié)合，可構(gòu)建更強大的緩存系統(tǒng)。
3. 異步支持
使用 asyncio.Lock 和 asyncio.sleep 實現(xiàn)異步版本，適用于異步框架（如 FastAPI）。
七、實戰(zhàn)案例：接口緩存中間層
在 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)) # 緩存命中
八、未來展望與生態(tài)融合
1. 與 Redis 結(jié)合
- 本地緩存命中失敗后，嘗試從 Redis 獲??；
- 本地緩存作為一級緩存，Redis 為二級緩存；
- 適用于分布式系統(tǒng)中的熱點數(shù)據(jù)加速。
2. 與 FastAPI / Flask 集成
- 將緩存作為中間件；
- 或封裝為依賴注入組件；
- 提升接口響應速度，降低數(shù)據(jù)庫壓力。
3. 與 Prometheus 結(jié)合
- 監(jiān)控緩存命中率、過期率、淘汰頻率；
- 提供可觀測性，輔助性能調(diào)優(yōu)。
九、總結(jié)與互動
我們從零實現(xiàn)了一個支持 TTL、并發(fā)安全、自動清理的本地緩存系統(tǒng)，并通過實戰(zhàn)驗證其在百萬級 QPS 場景下的性能表現(xiàn)。
這不僅是一次技術(shù)實現(xiàn)，更是一次系統(tǒng)設(shè)計思維的鍛煉。
開放問題：
- 你在實際項目中是否遇到過緩存穿透、雪崩等問題？是如何解決的？
- 如果讓你擴展這個緩存系統(tǒng)，你會加入哪些功能？（如分布式同步、異步支持、LRU、統(tǒng)計監(jiān)控等）
到此這篇關(guān)于從零帶你構(gòu)建Python高性能本地緩存系統(tǒng)的文章就介紹到這了,更多相關(guān)Python本地緩存系統(tǒng)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
python 裝飾器功能以及函數(shù)參數(shù)使用介紹
之前學習編程語言大多也就是學的很淺很淺，基本上也是很少涉及到裝飾器這些的類似的內(nèi)容?？偸怯X得是一樣很神奇的東西，舍不得學（嘿嘿）。今天看了一下書籍。發(fā)現(xiàn)道理還是很簡單的2012-01-01
這篇文章主要介紹了python(列表生成式/器)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-03-03
python常用的各種排序算法原理與實現(xiàn)方法小結(jié)
這篇文章主要介紹了python常用的各種排序算法原理與實現(xiàn)方法,結(jié)合實例形式總結(jié)分析了冒泡排序、插入排序、選擇排序、快速排序等排序算法的相關(guān)原理與實現(xiàn)方法,需要的朋友可以參考下2023-04-04
站在一個租房人的立場，租房平臺實在太多了，并且各平臺篩選和排序邏輯都不太一致。這篇文章將教教大家如何利用Python語言實現(xiàn)爬取房源信息，需要的可以參考一下2022-09-09
Python使用SymPy解決Manim曲線繪制速度不均的問題
這段文章詳細講解了使用SymPy進行弧長參數(shù)化以實現(xiàn)參數(shù)曲線均勻繪制的技術(shù),通過計算弧長函數(shù)、反解參數(shù)值及數(shù)值求解,實現(xiàn)曲線繪制節(jié)奏均勻,提升視覺體驗,關(guān)鍵代碼示例及效果展示進一步說明了方法的有效性,需要的朋友可以參考下2026-06-06
time庫是python中處理時間的標準庫,這篇文章主要介紹了Python中time庫的使用(日期時間),本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-02-02
Python系統(tǒng)監(jiān)控之跨平臺檢查系統(tǒng)服務運行狀態(tài)的完整指南
系統(tǒng)服務是操作系統(tǒng)和應用程序正常運行的基礎(chǔ),本文將和大家分享一個系統(tǒng)服務狀態(tài)檢查的Python實用腳本,可以跨平臺檢查系統(tǒng)服務的運行狀態(tài),并提供詳細的報告和告警功能,有需要的可以參考下2025-12-12
在使用conda創(chuàng)建指定目錄的Python環(huán)境時遇到錯誤,通過管理員身份運行、修改源配置、檢查環(huán)境目錄等多種方法,最終解決了問題,關(guān)鍵在于正確使用`conda?config?--append?envs_dirs`命令,指定環(huán)境所在目錄,而不是環(huán)境名稱2026-01-01
Python導入txt數(shù)據(jù)到mysql的方法
這篇文章主要介紹了Python導入txt數(shù)據(jù)到mysql的方法,涉及Python操作txt文件及mysql數(shù)據(jù)庫的技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-04-04
這篇文章主要介紹了Python Pivot table透視表使用方法解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-09-09