1. 從一段看起來能跑但內(nèi)存炸了的代碼說起很多人第一次接觸列表推導(dǎo)式都是被它那行云流水的寫法吸引的——一行頂五行簡潔、優(yōu)雅、Pythonic。但真正在生產(chǎn)環(huán)境里踩過坑的人都知道列表推導(dǎo)式、生成器、迭代器這三樣?xùn)|西表面上是語法糖和進(jìn)階特性實際上它們背后對應(yīng)的是三種完全不同的數(shù)據(jù)處理哲學(xué)一次性全量構(gòu)建、惰性按需產(chǎn)出、以及協(xié)議驅(qū)動的統(tǒng)一遍歷接口。搞不清楚這三者的邊界寫出來的代碼要么內(nèi)存爆掉要么邏輯詭異到調(diào)試半天找不到原因。我見過太多這樣的情況一個處理日志文件的腳本用列表推導(dǎo)式讀了幾十萬行數(shù)據(jù)本地測試沒問題一上服務(wù)器直接OOM或者有人寫了個生成器函數(shù)結(jié)果在里面用了return返回值調(diào)用方拿到的是個生成器對象而不是結(jié)果一臉懵。這些問題的根源都不是語法不熟而是對迭代協(xié)議和惰性求值的理解停留在表面。這篇內(nèi)容就是想把這三塊知識徹底串起來講清楚。不管你是剛學(xué)Python不久還是已經(jīng)寫了幾年代碼但對生成器始終會用但說不清我都會從底層機(jī)制、實際選型、性能對比、常見坑位幾個角度展開。核心關(guān)鍵詞就三個列表推導(dǎo)式、生成器、迭代器。讀完你應(yīng)該能做到看到一段數(shù)據(jù)處理代碼立刻判斷該用哪種方式以及為什么。2. 列表推導(dǎo)式的本質(zhì)它到底比for循環(huán)快在哪2.1 列表推導(dǎo)式的執(zhí)行模型先看一段最基礎(chǔ)的對比。假設(shè)我們要生成0到9的平方列表# 方式一普通for循環(huán) squares [] for i in range(10): squares.append(i ** 2) # 方式二列表推導(dǎo)式 squares [i ** 2 for i in range(10)]兩種寫法結(jié)果完全一樣但執(zhí)行過程有本質(zhì)區(qū)別。普通for循環(huán)每次迭代都要調(diào)用一次append方法這是一次方法查找加方法調(diào)用的開銷。而列表推導(dǎo)式在CPython層面被編譯成了一種專門的字節(jié)碼指令LIST_APPEND它直接操作列表對象跳過了屬性查找的環(huán)節(jié)。你可以用dis模塊自己驗證import dis def with_loop(): result [] for i in range(10): result.append(i ** 2) return result def with_comprehension(): return [i ** 2 for i in range(10)] dis.dis(with_loop) print(---) dis.dis(with_comprehension)實測下來在數(shù)據(jù)量較大的場景比如百萬級列表推導(dǎo)式通常比等價的for循環(huán)快20%到40%。這個差距在純計算密集型的任務(wù)里非常可觀。但要注意這個優(yōu)勢主要來自字節(jié)碼層面的優(yōu)化而不是推導(dǎo)式這個概念本身有什么魔法。2.2 條件過濾與嵌套推導(dǎo)式的正確姿勢列表推導(dǎo)式支持條件過濾語法是把if放在for后面# 只保留偶數(shù) evens [i for i in range(20) if i % 2 0] # 等價于 evens [] for i in range(20): if i % 2 0: evens.append(i)這里有個容易搞混的點(diǎn)if放在for前面和后面含義完全不同。放在后面是過濾放在前面是三元表達(dá)式# 過濾只保留偶數(shù)奇數(shù)直接丟棄 [x for x in range(10) if x % 2 0] # [0, 2, 4, 6, 8] # 三元所有元素都保留偶數(shù)原樣奇數(shù)變-1 [x if x % 2 0 else -1 for x in range(10)] # [0, -1, 2, -1, 4, ...]嵌套推導(dǎo)式也要小心順序。[x y for x in [1,2] for y in [10,20]]等價于兩層for循環(huán)外層是x內(nèi)層是y結(jié)果是[11, 21, 12, 22]。很多人第一次寫嵌套推導(dǎo)式會把順序搞反導(dǎo)致結(jié)果不對。2.3 列表推導(dǎo)式的內(nèi)存陷阱這是最需要警惕的地方。列表推導(dǎo)式會立即構(gòu)建完整的列表所有元素一次性放進(jìn)內(nèi)存。數(shù)據(jù)量小的時候無所謂但如果你處理的是文件行、數(shù)據(jù)庫記錄、網(wǎng)絡(luò)響應(yīng)這類可能很大的數(shù)據(jù)源列表推導(dǎo)式就是內(nèi)存殺手。# 危險寫法如果文件有1000萬行這個列表會吃掉大量內(nèi)存 lines [line.strip() for line in open(huge_file.txt)] # 安全寫法用生成器表達(dá)式逐行處理 lines (line.strip() for line in open(huge_file.txt))注意上面第二個寫法用的是圓括號而不是方括號這就是生成器表達(dá)式。它不會一次性構(gòu)建列表而是返回一個生成器對象每次迭代才產(chǎn)出一個值。內(nèi)存占用從O(n)降到O(1)。提示判斷該用列表推導(dǎo)式還是生成器表達(dá)式核心就一句話——這個結(jié)果后續(xù)需要被多次遍歷嗎需要索引訪問嗎需要知道長度嗎如果都不需要優(yōu)先用生成器。3. 生成器惰性求值背后的狀態(tài)機(jī)機(jī)制3.1 生成器函數(shù)與yield的執(zhí)行流程生成器函數(shù)就是包含yield關(guān)鍵字的函數(shù)。調(diào)用它不會執(zhí)行函數(shù)體而是返回一個生成器對象def count_up(n): print(開始執(zhí)行) i 0 while i n: yield i i 1 print(執(zhí)行結(jié)束) gen count_up(3) print(type(gen)) # class generator注意此時開始執(zhí)行并沒有被打印。只有當(dāng)你調(diào)用next(gen)或者用for循環(huán)遍歷時函數(shù)體才真正開始運(yùn)行。每次遇到y(tǒng)ield函數(shù)會暫停并交出當(dāng)前值同時保存所有局部變量的狀態(tài)。下次調(diào)用next時從暫停處繼續(xù)。這個暫停-恢復(fù)的能力底層是靠棧幀保留實現(xiàn)的。普通函數(shù)調(diào)用結(jié)束后棧幀就銷毀了而生成器函數(shù)的棧幀會被掛起保留直到生成器被耗盡或顯式關(guān)閉。這也是為什么生成器能記住i的值。gen count_up(3) print(next(gen)) # 打印開始執(zhí)行然后輸出0 print(next(gen)) # 輸出1 print(next(gen)) # 輸出2 print(next(gen)) # 打印執(zhí)行結(jié)束然后拋StopIteration3.2 生成器表達(dá)式與生成器函數(shù)的選型生成器表達(dá)式適合簡單的、一次性的惰性序列total sum(x ** 2 for x in range(1000000))生成器函數(shù)適合邏輯復(fù)雜、需要多步處理、或者需要復(fù)用的情況def read_large_file(path): with open(path) as f: for line in f: line line.strip() if line and not line.startswith(#): yield line選型原則如果邏輯能用一個表達(dá)式寫完用生成器表達(dá)式如果需要多行、有分支、有循環(huán)嵌套用生成器函數(shù)。不要為了炫技把復(fù)雜邏輯硬塞進(jìn)一個生成器表達(dá)式里可讀性會急劇下降。3.3 生成器的send、throw與close生成器不只是產(chǎn)出值它還支持雙向通信。send方法可以向生成器內(nèi)部傳值這個值會成為當(dāng)前yield表達(dá)式的返回值def accumulator(): total 0 while True: value yield total if value is None: break total value acc accumulator() next(acc) # 啟動生成器返回0 print(acc.send(10)) # 返回10 print(acc.send(20)) # 返回30throw用于在生成器內(nèi)部拋出異常close用于關(guān)閉生成器并觸發(fā)GeneratorExit。這些特性在協(xié)程、狀態(tài)機(jī)、管道處理等場景里非常有用但日常業(yè)務(wù)代碼用得不多。我的建議是先把基礎(chǔ)的yield和惰性求值吃透send/throw這些等真正需要時再深入。注意生成器只能被遍歷一次。遍歷完之后再調(diào)用next會直接拋StopIteration。如果你需要多次遍歷要么重新創(chuàng)建生成器要么把結(jié)果轉(zhuǎn)成列表但要考慮內(nèi)存。4. 迭代器協(xié)議for循環(huán)背后到底發(fā)生了什么4.1 可迭代對象與迭代器的區(qū)別這是面試高頻題也是實際寫代碼時最容易混淆的概念。先給結(jié)論可迭代對象Iterable實現(xiàn)了__iter__方法的對象能被iter()調(diào)用。迭代器Iterator同時實現(xiàn)了__iter__和__next__方法的對象能被next()調(diào)用。列表、元組、字典、字符串都是可迭代對象但它們不是迭代器。對它們調(diào)用iter()會返回一個迭代器nums [1, 2, 3] print(hasattr(nums, __iter__)) # True print(hasattr(nums, __next__)) # False it iter(nums) print(hasattr(it, __iter__)) # True print(hasattr(it, __next__)) # Truefor循環(huán)的本質(zhì)就是先調(diào)用iter()拿到迭代器然后不斷調(diào)用next()直到捕獲StopIteration異常為止。用while循環(huán)手動模擬就是nums [1, 2, 3] it iter(nums) while True: try: value next(it) except StopIteration: break print(value)4.2 自定義迭代器的完整實現(xiàn)理解了協(xié)議就可以自己寫迭代器。下面是一個斐波那契數(shù)列迭代器class Fibonacci: def __init__(self, limit): self.limit limit self.a 0 self.b 1 self.count 0 def __iter__(self): return self def __next__(self): if self.count self.limit: raise StopIteration value self.a self.a, self.b self.b, self.a self.b self.count 1 return value for num in Fibonacci(8): print(num) # 0 1 1 2 3 5 8 13關(guān)鍵點(diǎn)__iter__返回self說明迭代器本身也是可迭代的。__next__在耗盡時必須拋StopIteration這是協(xié)議約定不能返回None或者別的什么來表示結(jié)束。4.3 迭代器與生成器的關(guān)系生成器本質(zhì)上是一種語法糖形式的迭代器。任何生成器都自動實現(xiàn)了迭代器協(xié)議你不需要手動寫__iter__和__next__。反過來迭代器不一定非得是生成器上面那個Fibonacci類就是純手寫的迭代器。實際開發(fā)中90%的場景用生成器就夠了手寫迭代器類通常只在需要維護(hù)復(fù)雜狀態(tài)、或者需要提供額外方法比如reset時才用。我個人的經(jīng)驗是能用生成器函數(shù)解決的絕不手寫迭代器類因為生成器的狀態(tài)管理由解釋器負(fù)責(zé)不容易出錯。對比維度列表推導(dǎo)式生成器表達(dá)式生成器函數(shù)手寫迭代器類內(nèi)存占用O(n)O(1)O(1)O(1)可重復(fù)遍歷是否否取決于實現(xiàn)支持索引是否否否代碼復(fù)雜度低低中高適用場景小數(shù)據(jù)量、需多次訪問簡單惰性序列復(fù)雜惰性邏輯需要額外狀態(tài)管理5. 性能實測三種方式在不同數(shù)據(jù)規(guī)模下的表現(xiàn)5.1 內(nèi)存占用的量化對比光說生成器省內(nèi)存太抽象我們直接測。用sys.getsizeof看對象本身大小再用tracemalloc看實際內(nèi)存分配import tracemalloc N 1_000_000 # 列表推導(dǎo)式 tracemalloc.start() lst [i * 2 for i in range(N)] current, peak tracemalloc.get_traced_memory() tracemalloc.stop() print(f列表推導(dǎo)式峰值內(nèi)存: {peak / 1024 / 1024:.2f} MB) # 生成器表達(dá)式 tracemalloc.start() gen (i * 2 for i in range(N)) current, peak tracemalloc.get_traced_memory() tracemalloc.stop() print(f生成器表達(dá)式峰值內(nèi)存: {peak / 1024 / 1024:.2f} MB)在百萬級數(shù)據(jù)下列表推導(dǎo)式的峰值內(nèi)存通常在30到40MB左右取決于整數(shù)對象大小而生成器表達(dá)式幾乎可以忽略不計通常不到1MB。這個差距在千萬級、億級數(shù)據(jù)下會變成致命的區(qū)別。5.2 執(zhí)行速度的取舍但生成器不是沒有代價的。每次next調(diào)用都有額外的開銷所以在純速度上列表推導(dǎo)式通常更快import timeit # 列表推導(dǎo)式求和 t1 timeit.timeit([i * 2 for i in range(100000)], number100) # 生成器表達(dá)式求和 t2 timeit.timeit(sum(i * 2 for i in range(100000)), number100)實測結(jié)果通常是列表推導(dǎo)式快1.5到2倍。原因很簡單列表推導(dǎo)式在C層面批量構(gòu)建而生成器每次產(chǎn)出一個值都要經(jīng)過Python層面的狀態(tài)機(jī)調(diào)度。所以選型的核心權(quán)衡是內(nèi)存敏感還是速度敏感。如果數(shù)據(jù)量不大幾萬條以內(nèi)列表推導(dǎo)式更快更直接如果數(shù)據(jù)量大或者數(shù)據(jù)源是流式的文件、網(wǎng)絡(luò)、傳感器生成器是唯一合理的選擇。5.3 一個真實場景的選型推演假設(shè)你要處理一個日志文件統(tǒng)計所有ERROR級別的日志條數(shù)。文件可能很大但統(tǒng)計結(jié)果只是一個整數(shù)。錯誤做法lines [line for line in open(app.log)] # 全量加載內(nèi)存爆炸 error_count len([l for l in lines if ERROR in l])正確做法error_count sum(1 for line in open(app.log) if ERROR in line)第二種寫法全程只維護(hù)一個計數(shù)器內(nèi)存恒定。而且sum配合生成器表達(dá)式是Python里非常經(jīng)典的惰性統(tǒng)計模式值得記住。再進(jìn)一步如果還要按錯誤類型分組統(tǒng)計from collections import defaultdict stats defaultdict(int) with open(app.log) as f: for line in f: if ERROR in line: error_type line.split(])[0].strip([) stats[error_type] 1這里用普通for循環(huán)反而比推導(dǎo)式更清晰因為涉及多個操作。不要為了用推導(dǎo)式而用推導(dǎo)式這是很多人容易走偏的地方。6. 那些年我踩過的生成器與迭代器坑6.1 生成器只能消費(fèi)一次的隱蔽bug這是最經(jīng)典的坑?？催@段代碼def get_numbers(): for i in range(5): yield i nums get_numbers() print(list(nums)) # [0, 1, 2, 3, 4] print(list(nums)) # [] 空列表第二次list(nums)得到空列表因為生成器已經(jīng)耗盡了。這個bug在函數(shù)間傳遞生成器時特別隱蔽。比如你把生成器傳給一個函數(shù)做統(tǒng)計又傳給另一個函數(shù)做過濾第二個函數(shù)拿到的就是空的。解決方案有兩個要么在需要多次使用的地方轉(zhuǎn)成列表確認(rèn)內(nèi)存可控要么每次重新調(diào)用生成器函數(shù)創(chuàng)建新的生成器對象。我個人的習(xí)慣是生成器對象絕不跨函數(shù)邊界傳遞超過一次如果確實需要多處使用在源頭就明確是列表還是生成器。6.2 在生成器里使用return的返回值問題生成器函數(shù)里的return不會返回值給調(diào)用方而是觸發(fā)StopIterationdef gen(): yield 1 yield 2 return done g gen() print(next(g)) # 1 print(next(g)) # 2 try: next(g) except StopIteration as e: print(e.value) # donereturn的值藏在StopIteration異常的value屬性里。這個特性在yield from語法里有用但日常代碼里幾乎用不到。如果你在生成器里寫return result然后期望調(diào)用方拿到result那一定會失望。6.3 迭代過程中修改集合的災(zāi)難nums [1, 2, 3, 4, 5] for n in nums: if n % 2 0: nums.remove(n) print(nums) # [1, 3, 5]不一定在遍歷列表的同時修改它會導(dǎo)致迭代器索引錯亂結(jié)果不可預(yù)測。正確做法是遍歷副本或者用推導(dǎo)式重建nums [n for n in nums if n % 2 ! 0]字典和集合同理遍歷時不能增刪元素否則直接拋RuntimeError: dictionary changed size during iteration。這個坑在寫數(shù)據(jù)清洗邏輯時特別容易踩。6.4 生成器里的異常處理邊界生成器內(nèi)部的異常會在next調(diào)用時向外傳播。如果你在生成器里捕獲了異常但沒有正確處理可能導(dǎo)致生成器靜默終止def risky_gen(): for i in range(5): try: yield 10 / (i - 2) except ZeroDivisionError: yield 0 # 必須yield一個值否則這個位置就跳過了注意i2時除零如果不yield任何東西生成器會繼續(xù)下一個循環(huán)但調(diào)用方感知不到這里出過問題。是否要yield一個占位值取決于業(yè)務(wù)語義。我的建議是生成器內(nèi)部的異常要么向上拋要么明確產(chǎn)出一個錯誤標(biāo)記值不要靜默吞掉。7. 把三者串起來一個完整的數(shù)據(jù)管道實戰(zhàn)7.1 需求描述與整體設(shè)計假設(shè)我們要處理一個CSV格式的傳感器數(shù)據(jù)文件需求是讀取文件、過濾掉無效行、把溫度從攝氏度轉(zhuǎn)成華氏度、輸出前10條結(jié)果。文件可能很大不能全量加載。整體設(shè)計思路用生成器函數(shù)做文件讀取和過濾用生成器表達(dá)式做轉(zhuǎn)換最后用itertools.islice取前10條。整條管道全程惰性內(nèi)存恒定。import itertools def read_valid_rows(path): with open(path) as f: header next(f) # 跳過表頭 for line in f: parts line.strip().split(,) if len(parts) ! 3: continue try: sensor_id parts[0] temp_c float(parts[1]) timestamp parts[2] except ValueError: continue yield sensor_id, temp_c, timestamp def to_fahrenheit(rows): for sensor_id, temp_c, timestamp in rows: yield sensor_id, temp_c * 9 / 5 32, timestamp pipeline to_fahrenheit(read_valid_rows(sensor.csv)) first_ten list(itertools.islice(pipeline, 10))7.2 每個環(huán)節(jié)的選型理由read_valid_rows用生成器函數(shù)而不是列表推導(dǎo)式因為文件可能很大而且過濾邏輯涉及異常處理用推導(dǎo)式寫會很難看。to_fahrenheit也用生成器函數(shù)保持管道的惰性。最后的islice是關(guān)鍵——它只從管道里取10條意味著整個文件只會被讀取到滿足10條有效數(shù)據(jù)為止后面的內(nèi)容根本不會被處理。如果這里用列表推導(dǎo)式整個文件會被完整讀取和轉(zhuǎn)換然后才取前10條浪費(fèi)大量計算和內(nèi)存。這就是惰性管道的威力計算量與實際消費(fèi)量成正比而不是與數(shù)據(jù)源大小成正比。7.3 管道調(diào)試的實用技巧惰性管道有個調(diào)試難點(diǎn)出錯時不知道是哪一環(huán)的問題。我的做法是在每個環(huán)節(jié)臨時插入一個探針生成器def probe(rows, label): for row in rows: print(f[{label}] {row}) yield row pipeline to_fahrenheit(probe(read_valid_rows(sensor.csv), raw))這樣能看到數(shù)據(jù)在每個環(huán)節(jié)的形態(tài)。調(diào)試完把probe去掉即可。另一個技巧是用itertools.tee把一個生成器分成兩個一個用于調(diào)試輸出一個用于正常流程但要注意tee會緩存數(shù)據(jù)大流量下慎用。提示惰性管道最適合讀取-過濾-轉(zhuǎn)換-聚合這類線性流程。如果流程中有分支、需要回溯、或者需要多次遍歷同一批數(shù)據(jù)就該考慮在某個節(jié)點(diǎn)落地成列表或?qū)懭肱R時存儲。8. 關(guān)于這三塊知識我個人的幾條經(jīng)驗寫了這么多年P(guān)ython關(guān)于列表推導(dǎo)式、生成器、迭代器我總結(jié)了幾條不太會在教程里看到但很實用的經(jīng)驗。第一條列表推導(dǎo)式不要超過兩層嵌套。三層以上的嵌套推導(dǎo)式可讀性比等價的for循環(huán)差得多維護(hù)成本極高。我見過有人寫出四層嵌套的推導(dǎo)式同事看了十分鐘沒看懂最后重寫成for循環(huán)反而一目了然。簡潔不等于短可讀性永遠(yuǎn)優(yōu)先。第二條生成器的價值不只是省內(nèi)存更是解耦生產(chǎn)與消費(fèi)。當(dāng)你把數(shù)據(jù)生產(chǎn)邏輯寫成生成器消費(fèi)方可以按自己的節(jié)奏拉取兩者通過迭代協(xié)議解耦。這個思路在寫爬蟲、消息處理、流式計算時特別有用。生產(chǎn)者不需要知道消費(fèi)者什么時候要數(shù)據(jù)消費(fèi)者也不需要知道數(shù)據(jù)怎么來的。第三條迭代器協(xié)議是理解Python很多特性的鑰匙。for循環(huán)、解包、zip、map、filter、itertools、yield from、異步的async for全都建立在迭代協(xié)議之上。把__iter__和__next__這兩個方法吃透很多看似無關(guān)的知識點(diǎn)會自動串起來。第四條性能優(yōu)化先測量再動手。不要憑感覺認(rèn)為生成器一定比列表快或者慢。用timeit和tracemalloc實測用數(shù)據(jù)說話。我遇到過數(shù)據(jù)量只有幾千條卻非要用生成器的情況結(jié)果代碼復(fù)雜度上去了性能反而因為狀態(tài)機(jī)開銷略有下降。工具沒有絕對的好壞只有適不適合當(dāng)前場景。最后分享一個我常用的判斷口訣要多次用、要索引、數(shù)據(jù)小用列表只遍歷一次、數(shù)據(jù)大、流式來源用生成器需要自定義復(fù)雜狀態(tài)才手寫迭代器類。這三句話覆蓋了日常開發(fā)中絕大多數(shù)選型場景剩下的邊角情況等你真正遇到了自然會有判斷。