案例:LRU 快取
案例:LRU 快取
本案例基於 .claude/lib/git_utils.py 的 is_protected_branch() 和 is_allowed_branch() 函式,展示如何用 functools.lru_cache 快取重複的分支檢查結果。
先備知識
- 入門系列 3.8 效能優化
- 基本的 Git 分支概念
問題背景
現有設計
git_utils.py 提供兩個分支檢查函式:
1import fnmatch
2
3# 保護分支列表(支援 glob 模式)
4PROTECTED_BRANCHES = [
5 "main",
6 "master",
7 "develop",
8 "release/*",
9 "production",
10]
11
12# 允許編輯的分支模式
13ALLOWED_BRANCHES = [
14 "feat/*",
15 "feature/*",
16 "fix/*",
17 "hotfix/*",
18 "bugfix/*",
19 "chore/*",
20 "docs/*",
21 "refactor/*",
22 "test/*",
23]
24
25def is_protected_branch(branch: str) -> bool:
26 """
27 檢查是否為保護分支
28
29 Args:
30 branch: 分支名稱
31
32 Returns:
33 bool: 如果是保護分支返回 True
34 """
35 for pattern in PROTECTED_BRANCHES:
36 if fnmatch.fnmatch(branch, pattern):
37 return True
38 return False
39
40def is_allowed_branch(branch: str) -> bool:
41 """
42 檢查是否為允許編輯的分支
43
44 Args:
45 branch: 分支名稱
46
47 Returns:
48 bool: 如果是允許編輯的分支返回 True
49 """
50 for pattern in ALLOWED_BRANCHES:
51 if fnmatch.fnmatch(branch, pattern):
52 return True
53 return False這個設計的優點
- 簡單直覺:迴圈遍歷模式清單,易於理解
- 彈性高:支援 glob 模式,可輕易新增規則
- 無狀態:純函數,沒有副作用
重複計算的開銷
在 Hook 驗證流程中,同一個分支可能被檢查多次:
1def validate_hook(hook_config: dict) -> list[str]:
2 """驗證 Hook 配置"""
3 errors = []
4 branch = get_current_branch()
5
6 # 第一次檢查:是否允許執行
7 if is_protected_branch(branch):
8 errors.append("Cannot run on protected branch")
9
10 # ... 其他驗證邏輯 ...
11
12 # 第二次檢查:是否需要額外確認
13 if is_protected_branch(branch) and hook_config.get("dangerous"):
14 errors.append("Dangerous operation on protected branch")
15
16 return errors
17
18def process_multiple_hooks(hooks: list[dict]) -> None:
19 """處理多個 Hook"""
20 branch = get_current_branch()
21
22 for hook in hooks:
23 # 每個 Hook 都會檢查分支
24 if not is_allowed_branch(branch):
25 continue
26
27 # 再次檢查保護狀態
28 if is_protected_branch(branch):
29 # ... 特殊處理 ...
30 pass問題分析:
- 同一個分支名稱被重複檢查多次
- 每次檢查都要遍歷整個模式清單
fnmatch.fnmatch()雖然快,但重複呼叫仍是浪費
測量重複呼叫的影響:
1import time
2
3def benchmark_repeated_calls(branch: str, iterations: int = 10000) -> float:
4 """測量重複呼叫的時間"""
5 start = time.perf_counter()
6
7 for _ in range(iterations):
8 is_protected_branch(branch)
9 is_allowed_branch(branch)
10
11 return time.perf_counter() - start
12
13# 測量結果
14# 10000 次呼叫: ~0.05 秒
15# 每次呼叫: ~5 微秒
16
17# 看起來很快,但如果在熱路徑上...
18# 100 個 Hook x 10 次檢查 = 1000 次呼叫
19# 累積起來就有影響了進階解決方案
lru_cache 基礎
functools.lru_cache 是 Python 內建的記憶化(memoization)裝飾器:
1from functools import lru_cache
2
3@lru_cache(maxsize=128)
4def expensive_function(x: int) -> int:
5 """結果會被快取"""
6 return x * 2
7
8# 第一次呼叫:實際計算
9result1 = expensive_function(10) # 計算並快取
10
11# 第二次呼叫:直接返回快取結果
12result2 = expensive_function(10) # 從快取讀取lru_cache 的適用條件:
- 純函數:相同輸入永遠產生相同輸出
- 可雜湊參數:所有參數必須是 hashable(可作為 dict 的 key)
- 沒有副作用:函數不應修改外部狀態
- 計算成本 > 快取成本:快取有記憶體開銷,要值得
檢查 is_protected_branch 是否適用:
1# 1. 純函數?
2# 是 - 相同分支名稱永遠返回相同結果
3# (前提:PROTECTED_BRANCHES 不會在執行時改變)
4
5# 2. 可雜湊參數?
6# 是 - str 是 hashable
7
8# 3. 沒有副作用?
9# 是 - 只讀取全域常數,不修改任何狀態
10
11# 4. 計算成本值得快取?
12# 要測量...實作步驟
步驟 1:加入 lru_cache 裝飾器
1from functools import lru_cache
2import fnmatch
3
4PROTECTED_BRANCHES = [
5 "main",
6 "master",
7 "develop",
8 "release/*",
9 "production",
10]
11
12ALLOWED_BRANCHES = [
13 "feat/*",
14 "feature/*",
15 "fix/*",
16 "hotfix/*",
17 "bugfix/*",
18 "chore/*",
19 "docs/*",
20 "refactor/*",
21 "test/*",
22]
23
24@lru_cache(maxsize=128)
25def is_protected_branch(branch: str) -> bool:
26 """
27 檢查是否為保護分支(帶快取)
28
29 Args:
30 branch: 分支名稱
31
32 Returns:
33 bool: 如果是保護分支返回 True
34
35 Note:
36 結果會被快取,快取大小為 128 個不同的分支名稱。
37 如果 PROTECTED_BRANCHES 在執行時改變,需要呼叫
38 is_protected_branch.cache_clear() 清除快取。
39 """
40 for pattern in PROTECTED_BRANCHES:
41 if fnmatch.fnmatch(branch, pattern):
42 return True
43 return False
44
45@lru_cache(maxsize=128)
46def is_allowed_branch(branch: str) -> bool:
47 """
48 檢查是否為允許編輯的分支(帶快取)
49
50 Args:
51 branch: 分支名稱
52
53 Returns:
54 bool: 如果是允許編輯的分支返回 True
55
56 Note:
57 結果會被快取。修改 ALLOWED_BRANCHES 後需要
58 呼叫 is_allowed_branch.cache_clear()。
59 """
60 for pattern in ALLOWED_BRANCHES:
61 if fnmatch.fnmatch(branch, pattern):
62 return True
63 return False步驟 2:驗證快取行為
1def verify_cache_behavior():
2 """驗證快取確實在運作"""
3 # 清除快取,確保乾淨狀態
4 is_protected_branch.cache_clear()
5
6 # 第一次呼叫
7 result1 = is_protected_branch("main")
8 info1 = is_protected_branch.cache_info()
9 print(f"第一次呼叫: hits={info1.hits}, misses={info1.misses}")
10 # 輸出: hits=0, misses=1
11
12 # 第二次呼叫(相同參數)
13 result2 = is_protected_branch("main")
14 info2 = is_protected_branch.cache_info()
15 print(f"第二次呼叫: hits={info2.hits}, misses={info2.misses}")
16 # 輸出: hits=1, misses=1
17
18 # 不同參數
19 result3 = is_protected_branch("feature/new")
20 info3 = is_protected_branch.cache_info()
21 print(f"不同參數: hits={info3.hits}, misses={info3.misses}")
22 # 輸出: hits=1, misses=2
23
24verify_cache_behavior()maxsize 的選擇
maxsize 決定快取可以儲存多少個不同的輸入結果:
1# maxsize=None:無限制快取
2# 優點:所有結果都快取,命中率最高
3# 缺點:記憶體可能無限增長
4
5@lru_cache(maxsize=None)
6def unlimited_cache(x):
7 return x * 2
8
9# maxsize=128(預設):快取最多 128 個結果
10# LRU = Least Recently Used,超過時淘汰最久未使用的
11
12@lru_cache(maxsize=128)
13def limited_cache(x):
14 return x * 2
15
16# maxsize=1:只快取最後一個結果
17# 適用於「連續呼叫通常是相同參數」的情況
18
19@lru_cache(maxsize=1)
20def single_cache(x):
21 return x * 2選擇策略:
1def choose_maxsize():
2 """
3 選擇 maxsize 的考量因素
4 """
5 # 因素 1:輸入值的多樣性
6 # - 分支名稱通常不多(< 100 種)
7 # - 單一專案可能只有 10-20 個分支
8 # - maxsize=128 綽綽有餘
9
10 # 因素 2:記憶體使用
11 # - 每個快取項目: key + value + overhead
12 # - str 的 key 大小視長度而定
13 # - bool 的 value 很小
14 # - 128 個項目 < 10KB,可忽略
15
16 # 因素 3:呼叫模式
17 # - 如果同一個分支會被檢查很多次 → 大 maxsize
18 # - 如果每次都是新分支 → 快取沒意義
19
20 # 對於分支檢查:
21 # - 通常會重複檢查目前分支
22 # - maxsize=32 或 64 就足夠
23 # - 用 128 是保守選擇
24 pass實際測量 maxsize 的影響:
1from functools import lru_cache
2import random
3import time
4
5def benchmark_maxsize():
6 """比較不同 maxsize 的效能"""
7
8 # 模擬分支名稱
9 branches = [
10 "main", "master", "develop",
11 "feature/auth", "feature/api", "feature/ui",
12 "fix/bug1", "fix/bug2", "fix/bug3",
13 "release/v1.0", "release/v2.0",
14 ]
15
16 # 模擬呼叫模式:80% 是常見分支,20% 是其他
17 common_branches = branches[:3]
18
19 def simulate_calls(func, iterations=10000):
20 for _ in range(iterations):
21 if random.random() < 0.8:
22 branch = random.choice(common_branches)
23 else:
24 branch = random.choice(branches)
25 func(branch)
26 return func.cache_info()
27
28 # 測試不同 maxsize
29 for maxsize in [1, 4, 16, 64, 128, None]:
30 @lru_cache(maxsize=maxsize)
31 def test_func(branch):
32 for pattern in PROTECTED_BRANCHES:
33 if fnmatch.fnmatch(branch, pattern):
34 return True
35 return False
36
37 info = simulate_calls(test_func)
38 hit_rate = info.hits / (info.hits + info.misses) * 100
39
40 print(f"maxsize={str(maxsize):>4}: "
41 f"hit_rate={hit_rate:.1f}%, "
42 f"size={info.currsize}")
43
44# 輸出範例:
45# maxsize= 1: hit_rate=65.2%, size=1
46# maxsize= 4: hit_rate=92.8%, size=4
47# maxsize= 16: hit_rate=99.9%, size=11
48# maxsize= 64: hit_rate=99.9%, size=11
49# maxsize= 128: hit_rate=99.9%, size=11
50# maxsize=None: hit_rate=99.9%, size=11結論:對於分支檢查,maxsize=32 就足夠。用 maxsize=128 是安全的預設值。
快取命中率分析
cache_info() 提供詳細的快取統計:
1def analyze_cache_performance():
2 """分析快取效能"""
3 # 模擬一個 Hook 驗證流程
4 is_protected_branch.cache_clear()
5 is_allowed_branch.cache_clear()
6
7 # 模擬驗證 50 個 Hook,每個 Hook 檢查 2 次
8 branch = "feature/new-feature"
9
10 for _ in range(50):
11 # 每個 Hook 的驗證邏輯
12 is_protected_branch(branch)
13 is_allowed_branch(branch)
14
15 # 某些 Hook 會再次檢查
16 if not is_protected_branch(branch):
17 is_allowed_branch(branch)
18
19 # 分析結果
20 protected_info = is_protected_branch.cache_info()
21 allowed_info = is_allowed_branch.cache_info()
22
23 print("=== 快取效能分析 ===")
24 print(f"\nis_protected_branch:")
25 print(f" 命中: {protected_info.hits}")
26 print(f" 未命中: {protected_info.misses}")
27 print(f" 命中率: {protected_info.hits / (protected_info.hits + protected_info.misses) * 100:.1f}%")
28 print(f" 快取大小: {protected_info.currsize}/{protected_info.maxsize}")
29
30 print(f"\nis_allowed_branch:")
31 print(f" 命中: {allowed_info.hits}")
32 print(f" 未命中: {allowed_info.misses}")
33 print(f" 命中率: {allowed_info.hits / (allowed_info.hits + allowed_info.misses) * 100:.1f}%")
34 print(f" 快取大小: {allowed_info.currsize}/{allowed_info.maxsize}")
35
36# 輸出:
37# === 快取效能分析 ===
38#
39# is_protected_branch:
40# 命中: 99
41# 未命中: 1
42# 命中率: 99.0%
43# 快取大小: 1/128
44#
45# is_allowed_branch:
46# 命中: 99
47# 未命中: 1
48# 命中率: 99.0%
49# 快取大小: 1/128命中率解讀:
| 命中率 | 意義 | 建議 |
|---|---|---|
| > 90% | 快取非常有效 | 保持現狀 |
| 70-90% | 快取有幫助 | 可考慮增加 maxsize |
| 50-70% | 效果有限 | 檢查呼叫模式 |
| < 50% | 快取可能沒意義 | 考慮移除快取 |
完整程式碼
1#!/usr/bin/env python3
2"""
3分支檢查工具 - 帶 LRU 快取
4
5展示如何用 functools.lru_cache 優化重複的分支檢查。
6"""
7
8from functools import lru_cache
9import fnmatch
10from typing import Callable
11
12# ===== 分支配置常數 =====
13
14PROTECTED_BRANCHES = [
15 "main",
16 "master",
17 "develop",
18 "release/*",
19 "production",
20]
21
22ALLOWED_BRANCHES = [
23 "feat/*",
24 "feature/*",
25 "fix/*",
26 "hotfix/*",
27 "bugfix/*",
28 "chore/*",
29 "docs/*",
30 "refactor/*",
31 "test/*",
32]
33
34# ===== 帶快取的檢查函式 =====
35
36@lru_cache(maxsize=128)
37def is_protected_branch(branch: str) -> bool:
38 """
39 檢查是否為保護分支(帶快取)
40
41 Args:
42 branch: 分支名稱
43
44 Returns:
45 bool: 如果是保護分支返回 True
46
47 Example:
48 >>> is_protected_branch("main")
49 True
50 >>> is_protected_branch("feature/new")
51 False
52 """
53 for pattern in PROTECTED_BRANCHES:
54 if fnmatch.fnmatch(branch, pattern):
55 return True
56 return False
57
58@lru_cache(maxsize=128)
59def is_allowed_branch(branch: str) -> bool:
60 """
61 檢查是否為允許編輯的分支(帶快取)
62
63 Args:
64 branch: 分支名稱
65
66 Returns:
67 bool: 如果是允許編輯的分支返回 True
68
69 Example:
70 >>> is_allowed_branch("feat/new-feature")
71 True
72 >>> is_allowed_branch("random-branch")
73 False
74 """
75 for pattern in ALLOWED_BRANCHES:
76 if fnmatch.fnmatch(branch, pattern):
77 return True
78 return False
79
80# ===== 快取管理工具 =====
81
82def clear_branch_caches() -> None:
83 """
84 清除所有分支檢查快取
85
86 當 PROTECTED_BRANCHES 或 ALLOWED_BRANCHES 在執行時
87 被修改時,需要呼叫此函式。
88
89 Example:
90 >>> PROTECTED_BRANCHES.append("staging")
91 >>> clear_branch_caches() # 清除舊的快取結果
92 """
93 is_protected_branch.cache_clear()
94 is_allowed_branch.cache_clear()
95
96def get_cache_stats() -> dict:
97 """
98 獲取快取統計資訊
99
100 Returns:
101 dict: 包含兩個函式的快取統計
102
103 Example:
104 >>> stats = get_cache_stats()
105 >>> print(f"protected hit rate: {stats['protected']['hit_rate']:.1f}%")
106 """
107 protected = is_protected_branch.cache_info()
108 allowed = is_allowed_branch.cache_info()
109
110 def calc_hit_rate(info):
111 total = info.hits + info.misses
112 return info.hits / total * 100 if total > 0 else 0.0
113
114 return {
115 "protected": {
116 "hits": protected.hits,
117 "misses": protected.misses,
118 "hit_rate": calc_hit_rate(protected),
119 "size": protected.currsize,
120 "maxsize": protected.maxsize,
121 },
122 "allowed": {
123 "hits": allowed.hits,
124 "misses": allowed.misses,
125 "hit_rate": calc_hit_rate(allowed),
126 "size": allowed.currsize,
127 "maxsize": allowed.maxsize,
128 },
129 }
130
131def print_cache_stats() -> None:
132 """印出快取統計的格式化報告"""
133 stats = get_cache_stats()
134
135 print("=== 分支檢查快取統計 ===")
136
137 for name, info in stats.items():
138 print(f"\n{name}:")
139 print(f" 命中: {info['hits']}, 未命中: {info['misses']}")
140 print(f" 命中率: {info['hit_rate']:.1f}%")
141 print(f" 快取大小: {info['size']}/{info['maxsize']}")
142
143# ===== 效能比較工具 =====
144
145def benchmark_with_without_cache(
146 branches: list[str],
147 iterations: int = 1000
148) -> dict:
149 """
150 比較有無快取的效能差異
151
152 Args:
153 branches: 要測試的分支列表
154 iterations: 每個分支的呼叫次數
155
156 Returns:
157 dict: 效能比較結果
158 """
159 import time
160
161 # 無快取版本
162 def is_protected_no_cache(branch: str) -> bool:
163 for pattern in PROTECTED_BRANCHES:
164 if fnmatch.fnmatch(branch, pattern):
165 return True
166 return False
167
168 # 測試無快取版本
169 start = time.perf_counter()
170 for branch in branches:
171 for _ in range(iterations):
172 is_protected_no_cache(branch)
173 no_cache_time = time.perf_counter() - start
174
175 # 清除快取,測試有快取版本
176 is_protected_branch.cache_clear()
177
178 start = time.perf_counter()
179 for branch in branches:
180 for _ in range(iterations):
181 is_protected_branch(branch)
182 with_cache_time = time.perf_counter() - start
183
184 return {
185 "no_cache_time": no_cache_time,
186 "with_cache_time": with_cache_time,
187 "speedup": no_cache_time / with_cache_time,
188 "cache_info": is_protected_branch.cache_info(),
189 }
190
191# ===== 示範 =====
192
193if __name__ == "__main__":
194 import random
195
196 print("=== LRU 快取示範 ===\n")
197
198 # 測試分支
199 test_branches = [
200 "main",
201 "feature/auth",
202 "fix/bug-123",
203 "develop",
204 "chore/cleanup",
205 "release/v1.0",
206 ]
207
208 # 模擬重複呼叫
209 print("1. 模擬 Hook 驗證流程(100 次迭代):")
210 clear_branch_caches()
211
212 for _ in range(100):
213 branch = random.choice(test_branches)
214 is_protected_branch(branch)
215 is_allowed_branch(branch)
216
217 print_cache_stats()
218
219 # 效能比較
220 print("\n2. 效能比較:")
221 result = benchmark_with_without_cache(test_branches, iterations=10000)
222
223 print(f" 無快取: {result['no_cache_time']:.4f}s")
224 print(f" 有快取: {result['with_cache_time']:.4f}s")
225 print(f" 加速比: {result['speedup']:.1f}x")
226 print(f" 快取命中率: {result['cache_info'].hits / (result['cache_info'].hits + result['cache_info'].misses) * 100:.1f}%")設計權衡
| 面向 | 無快取 | lru_cache |
|---|---|---|
| 記憶體使用 | 無額外開銷 | 每個不同輸入一個快取項目 |
| 首次呼叫 | 直接計算 | 計算 + 快取開銷 |
| 重複呼叫 | 每次都計算 | O(1) 查表 |
| 程式碼複雜度 | 最簡單 | 加一行裝飾器 |
| 正確性風險 | 無 | 配置變更時需清除快取 |
| 可測試性 | 直接測試 | 需考慮快取狀態 |
| 執行緒安全 | 是 | 是(lru_cache 內建鎖) |
何時不該用快取
情況 1:函數不是純函數
1import os
2
3# 錯誤示範:結果依賴外部狀態
4@lru_cache(maxsize=128)
5def get_env_value(key: str) -> str:
6 return os.environ.get(key, "")
7
8# 問題:環境變數改變後,快取還是返回舊值
9os.environ["MY_VAR"] = "old"
10print(get_env_value("MY_VAR")) # "old"
11
12os.environ["MY_VAR"] = "new"
13print(get_env_value("MY_VAR")) # 仍然是 "old"!情況 2:參數不可雜湊
1# 錯誤示範:list 不是 hashable
2@lru_cache(maxsize=128)
3def process_items(items: list) -> int: # TypeError!
4 return sum(items)
5
6# 解法:用 tuple 代替 list
7@lru_cache(maxsize=128)
8def process_items(items: tuple) -> int:
9 return sum(items)情況 3:計算太簡單
1# 不建議:快取開銷可能大於計算成本
2@lru_cache(maxsize=128)
3def is_empty(s: str) -> bool:
4 return len(s) == 0
5
6# len() 和 == 操作非常快
7# 快取的雜湊計算和查表可能更慢情況 4:輸入值極度多樣
1# 不建議:每次輸入都不同,快取永遠不會命中
2@lru_cache(maxsize=128)
3def process_unique_id(unique_id: str) -> dict:
4 # 如果 unique_id 每次都不同...
5 return {"id": unique_id}
6
7# 快取只會一直 miss,浪費記憶體情況 5:需要即時反映外部變化
1# 不建議:配置可能動態變化
2@lru_cache(maxsize=128)
3def is_feature_enabled(feature: str) -> bool:
4 # 從資料庫讀取 feature flag
5 return db.get_feature_flag(feature)
6
7# 問題:資料庫更新後,快取還是返回舊值
8# 需要額外的快取失效機制什麼時候該用這個技術?
適合使用 lru_cache:
- 純函數(相同輸入,相同輸出)
- 會被重複呼叫,且輸入值有限
- 計算成本明顯高於雜湊和查表
- 配置是靜態的,不會在執行時改變
不適合使用:
- 函數有副作用或依賴外部狀態
- 參數不可雜湊(list、dict、set)
- 每次輸入都不同(如 UUID、時間戳)
- 計算非常簡單(如
len()、+)
快速決策流程:
1函數是純函數嗎?
2├── 否 → 不要用 lru_cache
3└── 是 → 參數都是 hashable 嗎?
4 ├── 否 → 轉換成 hashable(如 tuple)或不用
5 └── 是 → 會被重複呼叫嗎?
6 ├── 否 → 不需要快取
7 └── 是 → 適合用 lru_cache練習
基礎練習
測量快取效益:在你的專案中找一個被重複呼叫的純函數,加入
lru_cache,比較效能差異。監控快取狀態:寫一個裝飾器,在每次呼叫後印出
cache_info(),觀察快取行為。
進階練習
- 帶 TTL 的快取:實作一個有過期時間的快取裝飾器。提示:可以結合
time.time()和lru_cache。
1from functools import lru_cache, wraps
2import time
3
4def timed_lru_cache(seconds: int, maxsize: int = 128):
5 """
6 帶過期時間的 LRU 快取
7
8 Args:
9 seconds: 快取過期秒數
10 maxsize: 快取大小
11
12 用法:
13 @timed_lru_cache(seconds=60)
14 def fetch_data(url):
15 ...
16 """
17 def decorator(func):
18 # 你的實作
19 pass
20 return decorator- 快取命中率監控:建立一個監控系統,追蹤多個快取函式的命中率,當命中率低於閾值時發出警告。
挑戰題
- 動態配置的快取失效:修改
is_protected_branch,支援在執行時新增保護分支,並自動失效相關快取(而不是清除所有快取)。
提示:考慮用 typed=True 選項,或自訂快取類別。