問題背景
在高效能系統架構中，快取策略的選擇和實作直接影響系統效能。傳統的單一 LRU 實作往往缺乏擴展性，當業務需求變化時（如需要支援 LFU、FIFO 等策略），往往需要重寫整個快取模組。
OAVOService 面試助攻團隊 在輔導 Google、Amazon、Meta 等大廠系統設計面試中發現，可擴展快取設計是考察候選人物件導向設計能力和架構思維的經典題目。
問題描述
Problem 2 — Design an Extensible LRU Cache
設計一個可擴展的 LRU 快取類別結構並實作。重點考慮擴展性，因為後續需要新增更多淘汰策略，如 LFU（Least Frequently Used）。
核心要求：
- 實作基礎 LRU 快取功能（O(1) get/put 操作）
- 設計支援多種淘汰策略的可擴展架構
- 考慮未來新增 LFU、FIFO、Random 等策略
- 良好的介面抽象和程式碼復用
架構設計
1. 策略模式 + 工廠模式
from abc import ABC, abstractmethod
from typing import Any, Optional
import threading
class EvictionStrategy(ABC):
"""淘汰策略抽象基底類別"""
@abstractmethod
def on_access(self, key: Any) -> None:
"""鍵被存取時的回呼"""
pass
@abstractmethod
def on_insert(self, key: Any, value: Any) -> None:
"""插入新鍵值對時的回呼"""
pass
@abstractmethod
def evict(self) -> Any:
"""回傳需要淘汰的鍵"""
pass
@abstractmethod
def remove(self, key: Any) -> None:
"""移除指定鍵"""
pass
@abstractmethod
def size(self) -> int:
"""回傳目前大小"""
pass
class CacheNode:
"""雙向鏈表節點"""
def __init__(self, key=None, value=None):
self.key = key
self.value = value
self.prev = None
self.next = None
self.frequency = 1 # 為LFU策略預留
self.timestamp = 0 # 為其他策略預留
2. LRU 策略實作
import time
class LRUStrategy(EvictionStrategy):
"""LRU淘汰策略實作"""
def __init__(self):
# 雙向鏈表哨兵節點
self.head = CacheNode()
self.tail = CacheNode()
self.head.next = self.tail
self.tail.prev = self.head
self.key_to_node = {}
def _add_to_head(self, node: CacheNode) -> None:
"""新增節點到頭部"""
node.prev = self.head
node.next = self.head.next
self.head.next.prev = node
self.head.next = node
def _remove_node(self, node: CacheNode) -> None:
"""移除指定節點"""
node.prev.next = node.next
node.next.prev = node.prev
def _move_to_head(self, node: CacheNode) -> None:
"""移動節點到頭部"""
self._remove_node(node)
self._add_to_head(node)
def on_access(self, key: Any) -> None:
"""LRU存取：移動到頭部"""
if key in self.key_to_node:
node = self.key_to_node[key]
self._move_to_head(node)
def on_insert(self, key: Any, value: Any) -> None:
"""LRU插入：新增到頭部"""
node = CacheNode(key, value)
self.key_to_node[key] = node
self._add_to_head(node)
def evict(self) -> Any:
"""LRU淘汰：移除尾部節點"""
last_node = self.tail.prev
if last_node != self.head:
self._remove_node(last_node)
del self.key_to_node[last_node.key]
return last_node.key
return None
def remove(self, key: Any) -> None:
"""移除指定鍵"""
if key in self.key_to_node:
node = self.key_to_node[key]
self._remove_node(node)
del self.key_to_node[key]
def size(self) -> int:
return len(self.key_to_node)
def get_node(self, key: Any) -> Optional[CacheNode]:
return self.key_to_node.get(key)
3. 核心快取類別
class ExtensibleCache:
"""可擴展快取核心類別"""
def __init__(self, capacity: int, strategy: EvictionStrategy):
self.capacity = capacity
self.strategy = strategy
self._lock = threading.RLock() # 執行緒安全
def get(self, key: Any) -> Any:
"""取得快取值 O(1)"""
with self._lock:
node = self.strategy.get_node(key) if hasattr(self.strategy, 'get_node') else None
if node:
self.strategy.on_access(key)
return node.value
return None
def put(self, key: Any, value: Any) -> None:
"""設定快取值 O(1)"""
with self._lock:
# 檢查是否已存在
existing_node = self.strategy.get_node(key) if hasattr(self.strategy, 'get_node') else None
if existing_node:
existing_node.value = value
self.strategy.on_access(key)
return
# 檢查容量限制
if self.strategy.size() >= self.capacity:
evicted_key = self.strategy.evict()
if evicted_key is None:
raise RuntimeError("Failed to evict item from full cache")
# 插入新項目
self.strategy.on_insert(key, value)
def remove(self, key: Any) -> bool:
"""移除指定鍵"""
with self._lock:
if hasattr(self.strategy, 'get_node') and self.strategy.get_node(key):
self.strategy.remove(key)
return True
return False
def size(self) -> int:
"""目前快取大小"""
return self.strategy.size()
def clear(self) -> None:
"""清空快取"""
with self._lock:
# 重新初始化策略
self.strategy = type(self.strategy)()
擴展策略實作
1. LFU 策略擴展
from collections import defaultdict
import heapq
class LFUStrategy(EvictionStrategy):
"""LFU (Least Frequently Used) 策略"""
def __init__(self):
self.key_to_node = {}
self.freq_to_keys = defaultdict(set)
self.min_frequency = 0
self.key_to_freq = {}
def _update_frequency(self, key: Any) -> None:
"""更新鍵的頻率"""
freq = self.key_to_freq.get(key, 0)
# 從舊頻率組中移除
if freq > 0:
self.freq_to_keys[freq].discard(key)
if freq == self.min_frequency and not self.freq_to_keys[freq]:
self.min_frequency += 1
# 新增到新頻率組
new_freq = freq + 1
self.key_to_freq[key] = new_freq
self.freq_to_keys[new_freq].add(key)
def on_access(self, key: Any) -> None:
if key in self.key_to_node:
self._update_frequency(key)
def on_insert(self, key: Any, value: Any) -> None:
node = CacheNode(key, value)
self.key_to_node[key] = node
self.key_to_freq[key] = 1
self.freq_to_keys[1].add(key)
self.min_frequency = 1
def evict(self) -> Any:
if not self.key_to_node:
return None
# 找到最低頻率的鍵
evict_key = self.freq_to_keys[self.min_frequency].pop()
del self.key_to_node[evict_key]
del self.key_to_freq[evict_key]
return evict_key
def remove(self, key: Any) -> None:
if key in self.key_to_node:
freq = self.key_to_freq[key]
self.freq_to_keys[freq].discard(key)
del self.key_to_node[key]
del self.key_to_freq[key]
def size(self) -> int:
return len(self.key_to_node)
def get_node(self, key: Any) -> Optional[CacheNode]:
return self.key_to_node.get(key)
2. 快取工廠
class CacheFactory:
"""快取工廠類別"""
@staticmethod
def create_cache(cache_type: str, capacity: int) -> ExtensibleCache:
"""建立指定類型的快取"""
strategies = {
'lru': LRUStrategy,
'lfu': LFUStrategy,
'fifo': FIFOStrategy, # 可繼續擴展
'random': RandomStrategy,
}
if cache_type.lower() not in strategies:
raise ValueError(f"Unsupported cache type: {cache_type}")
strategy = strategies[cache_type.lower()]()
return ExtensibleCache(capacity, strategy)
# 使用範例
def demo_usage():
# 建立LRU快取
lru_cache = CacheFactory.create_cache('lru', 3)
lru_cache.put('a', 1)
lru_cache.put('b', 2)
lru_cache.put('c', 3)
print(lru_cache.get('a')) # 1, 'a'被存取
lru_cache.put('d', 4) # 淘汰'b'
print(lru_cache.get('b')) # None
# 建立LFU快取
lfu_cache = CacheFactory.create_cache('lfu', 2)
lfu_cache.put('x', 10)
lfu_cache.put('y', 20)
lfu_cache.get('x') # x頻率增加
lfu_cache.put('z', 30) # 淘汰y (頻率最低)
進階功能
1. 過期時間支援
class TTLExtension:
"""TTL (Time To Live) 擴展"""
def __init__(self, base_strategy: EvictionStrategy):
self.base_strategy = base_strategy
self.ttl_map = {} # key -> expiry_time
def set_ttl(self, key: Any, ttl_seconds: int) -> None:
expiry_time = time.time() + ttl_seconds
self.ttl_map[key] = expiry_time
def is_expired(self, key: Any) -> bool:
if key not in self.ttl_map:
return False
return time.time() > self.ttl_map[key]
def cleanup_expired(self) -> None:
current_time = time.time()
expired_keys = [k for k, v in self.ttl_map.items() if current_time > v]
for key in expired_keys:
self.base_strategy.remove(key)
del self.ttl_map[key]
2. 統計監控
class CacheStatistics:
"""快取統計資訊"""
def __init__(self):
self.hits = 0
self.misses = 0
self.evictions = 0
self.puts = 0
def record_hit(self):
self.hits += 1
def record_miss(self):
self.misses += 1
def record_eviction(self):
self.evictions += 1
def record_put(self):
self.puts += 1
@property
def hit_rate(self) -> float:
total = self.hits + self.misses
return self.hits / total if total > 0 else 0.0
OAVOService 面試技巧
設計原則討論
- 單一職責 - 策略類別只負責淘汰邏輯
- 開閉原則 - 對擴展開放，對修改封閉
- 相依反轉 - 相依抽象而非具體實作
常見面試擴展
- 分散式快取一致性
- 記憶體使用優化策略
- 並行存取效能優化
- 持久化和復原機制
複雜度分析
- LRU操作: O(1) get/put/remove
- LFU操作: O(1) get/put, O(log k) remove (k為不同頻率數)
- 空間複雜度: O(capacity)
OAVOService 專業面試助攻服務 - 深耕系統設計面試多年，為學員提供 Google、Amazon、Meta、Citadel 等頂級公司的即時面試輔助。我們的專家團隊精通各類設計模式和架構原理，確保你在系統設計面試中展現專業水準。
了解我們的 客製化系統設計面試方案 —— 從快取架構到分散式系統，全方位提升你的設計能力和面試表現。
標籤: LRU快取, 系統設計, 設計模式, 物件導向, 架構設計, OAVOService