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