用：從yield到內(nèi)存優(yōu)化)
1. 從函數(shù)到生成器的跨越第一次遇到y(tǒng)ield關(guān)鍵字時(shí)我正嘗試處理一個(gè)超過(guò)10GB的日志文件。傳統(tǒng)方法是將整個(gè)文件讀入內(nèi)存結(jié)果自然是內(nèi)存溢出。同事建議我試試生成器從此打開了新世界的大門。yield是Python中一個(gè)神奇的關(guān)鍵字它能讓普通函數(shù)搖身一變成為生成器函數(shù)。與return不同yield會(huì)暫停函數(shù)執(zhí)行并記住當(dāng)前位置的狀態(tài)下次調(diào)用時(shí)從斷點(diǎn)繼續(xù)執(zhí)行。這種特性在內(nèi)存敏感型應(yīng)用中大放異彩。def read_large_file(file_path): with open(file_path) as f: while True: line f.readline() if not line: break yield line這個(gè)簡(jiǎn)單的生成器函數(shù)完美解決了我的內(nèi)存問(wèn)題。它每次只讀取一行到內(nèi)存通過(guò)yield逐行返回而不是一次性加載整個(gè)文件。這就是生成器的核心價(jià)值——按需生成數(shù)據(jù)避免不必要的內(nèi)存消耗。2. 生成器的工作原理剖析2.1 生成器函數(shù)的執(zhí)行流程當(dāng)Python解釋器遇到包含yield的函數(shù)時(shí)不會(huì)像普通函數(shù)那樣立即執(zhí)行而是返回一個(gè)生成器對(duì)象。這個(gè)對(duì)象實(shí)現(xiàn)了迭代器協(xié)議包含以下幾個(gè)關(guān)鍵狀態(tài)創(chuàng)建階段調(diào)用生成器函數(shù)返回生成器對(duì)象此時(shí)代碼尚未執(zhí)行預(yù)激階段首次調(diào)用next()時(shí)代碼執(zhí)行到第一個(gè)yield處暫停掛起階段yield返回右側(cè)表達(dá)式結(jié)果保存所有局部變量狀態(tài)恢復(fù)階段再次調(diào)用next()時(shí)從上次暫停處繼續(xù)執(zhí)行def countdown(n): print(Starting countdown!) while n 0: yield n n - 1 print(Blastoff!) # 創(chuàng)建生成器 counter countdown(3) print(next(counter)) # 輸出: Starting countdown! 然后 3 print(next(counter)) # 輸出 2 print(next(counter)) # 輸出 1 print(next(counter)) # 輸出 Blastoff! 然后拋出StopIteration2.2 生成器的內(nèi)存模型生成器最顯著的優(yōu)勢(shì)是內(nèi)存效率。對(duì)比以下兩種實(shí)現(xiàn)# 傳統(tǒng)列表方式 def squares_list(n): result [] for i in range(n): result.append(i*i) return result # 生成器方式 def squares_gen(n): for i in range(n): yield i*i當(dāng)n1,000,000時(shí)列表版本需要存儲(chǔ)所有計(jì)算結(jié)果內(nèi)存占用約8MB假設(shè)每個(gè)整數(shù)8字節(jié)。而生成器版本在任何時(shí)候只維護(hù)當(dāng)前迭代狀態(tài)內(nèi)存占用幾乎可以忽略不計(jì)。3. yield的高級(jí)用法3.1 生成器表達(dá)式Python提供了更簡(jiǎn)潔的生成器表達(dá)式語(yǔ)法類似于列表推導(dǎo)式# 列表推導(dǎo)式 squares_list [x*x for x in range(10)] # 生成器表達(dá)式 squares_gen (x*x for x in range(10))生成器表達(dá)式特別適合處理大數(shù)據(jù)流比如統(tǒng)計(jì)大型文件中滿足條件的行數(shù)matched_lines sum(1 for line in open(huge.log) if error in line)3.2 yield from語(yǔ)法Python 3.3引入的yield from語(yǔ)法進(jìn)一步簡(jiǎn)化了生成器的嵌套使用def chain(*iterables): for it in iterables: yield from it # 等價(jià)于 def chain_manual(*iterables): for it in iterables: for item in it: yield itemyield from不僅能簡(jiǎn)化代碼還能保持子生成器的返回值def subgenerator(): yield 1 yield 2 return Done def delegator(): result yield from subgenerator() print(fSubgenerator returned: {result}) list(delegator()) # 輸出: Subgenerator returned: Done3.3 協(xié)程與雙向通信生成器通過(guò)send()方法實(shí)現(xiàn)了雙向通信這是協(xié)程的基礎(chǔ)def coroutine(): print(Starting coroutine) while True: received yield print(fReceived: {received}) c coroutine() next(c) # 預(yù)激生成器 c.send(Hello) # 輸出: Received: Hello c.send(World) # 輸出: Received: World這種模式在異步編程中非常有用雖然現(xiàn)代Python更推薦使用async/await語(yǔ)法但理解其底層原理很有必要。4. 實(shí)戰(zhàn)中的陷阱與技巧4.1 生成器只能消費(fèi)一次新手常犯的錯(cuò)誤是重復(fù)使用已耗盡的生成器gen (x for x in range(3)) print(list(gen)) # [0, 1, 2] print(list(gen)) # [] 第二次為空解決方案是重新創(chuàng)建生成器或者使用itertools.tee進(jìn)行復(fù)制注意內(nèi)存開銷。4.2 預(yù)激生成器的必要性需要接收數(shù)據(jù)的生成器必須先調(diào)用next()或send(None)進(jìn)行預(yù)激def echo(): while True: received yield print(received) e echo() e.send(hello) # 報(bào)錯(cuò): cant send non-None value to a just-started generator next(e) # 預(yù)激 e.send(hello) # 正常輸出: hello4.3 性能優(yōu)化技巧雖然生成器節(jié)省內(nèi)存但調(diào)用開銷比列表迭代大。對(duì)于小數(shù)據(jù)集直接使用列表可能更快# 測(cè)試代碼 import timeit small_data range(100) large_data range(1000000) def test_list(data): return [x*x for x in data] def test_gen(data): return list(x*x for x in data) # 小數(shù)據(jù)測(cè)試 print(timeit.timeit(lambda: test_list(small_data), number10000)) # 約0.3秒 print(timeit.timeit(lambda: test_gen(small_data), number10000)) # 約0.4秒 # 大數(shù)據(jù)測(cè)試 print(timeit.timeit(lambda: test_list(large_data), number1)) # 約0.5秒高內(nèi)存 print(timeit.timeit(lambda: test_gen(large_data), number1)) # 約0.6秒低內(nèi)存4.4 調(diào)試生成器調(diào)試生成器可能比較棘手因?yàn)閳?zhí)行流程不是線性的。我常用的方法是添加打印語(yǔ)句def debug_gen(): for i in range(3): print(fYielding {i}) yield i print(fResumed after {i})使用Python 3.7的breakpoint()def debug_gen(): for i in range(3): breakpoint() # 進(jìn)入pdb調(diào)試器 yield i將生成器轉(zhuǎn)換為列表查看所有值注意內(nèi)存消耗gen some_generator() print(list(gen)) # 查看所有輸出5. 生成器在標(biāo)準(zhǔn)庫(kù)中的應(yīng)用Python標(biāo)準(zhǔn)庫(kù)中大量使用了生成器模式典型例子包括5.1 itertools模塊itertools提供了豐富的生成器工具import itertools # 無(wú)限計(jì)數(shù)器 counter itertools.count(start10, step2) print(next(counter)) # 10 print(next(counter)) # 12 # 排列組合 perms itertools.permutations(ABC, 2) print(list(perms)) # [(A, B), (A, C), (B, A), ...] # 分組操作 groups itertools.groupby(AAABBBCCAAA) print([(k, list(g)) for k, g in groups]) # [(A, [A, A, A]), ...]5.2 上下文管理器contextlib.contextmanager裝飾器可以用生成器實(shí)現(xiàn)上下文管理器from contextlib import contextmanager contextmanager def timed_block(label): start time.time() try: yield finally: end time.time() print(f{label} took {end-start:.2f} seconds) with timed_block(calculation): time.sleep(1) # 模擬耗時(shí)操作5.3 文件處理csv模塊的reader函數(shù)返回生成器避免一次性加載大文件import csv def process_large_csv(filepath): with open(filepath) as f: reader csv.reader(f) for row in reader: yield process_row(row) # 逐行處理6. 生成器與異步編程雖然現(xiàn)代Python使用async/await語(yǔ)法處理協(xié)程但理解其與生成器的關(guān)系很有幫助6.1 歷史演變Python異步編程經(jīng)歷了多個(gè)階段生成器yield/sendPython 2.5asyncio.coroutineyield fromPython 3.4async/awaitPython 3.56.2 底層相似性async/await本質(zhì)上是生成器語(yǔ)法糖# 傳統(tǒng)生成器協(xié)程 def old_coroutine(): yield from asyncio.sleep(1) return 42 # 現(xiàn)代async協(xié)程 async def new_coroutine(): await asyncio.sleep(1) return 426.3 實(shí)際應(yīng)用案例生成器非常適合實(shí)現(xiàn)簡(jiǎn)單的狀態(tài)機(jī)def traffic_light(): while True: yield red yield green yield yellow light traffic_light() print(next(light)) # red print(next(light)) # green print(next(light)) # yellow另一個(gè)實(shí)用案例是分塊處理數(shù)據(jù)def chunker(iterable, size): for i in range(0, len(iterable), size): yield iterable[i:isize] for chunk in chunker(range(100), 10): process(chunk) # 每次處理10個(gè)元素7. 性能對(duì)比與最佳實(shí)踐7.1 生成器vs列表的內(nèi)存對(duì)比通過(guò)memory_profiler實(shí)測(cè)內(nèi)存使用profile def list_version(): return [i*i for i in range(1000000)] profile def gen_version(): return (i*i for i in range(1000000)) list_version() # 內(nèi)存峰值約40MB gen_version() # 內(nèi)存峰值基本不變7.2 何時(shí)使用生成器推薦使用生成器的場(chǎng)景處理大型或無(wú)限數(shù)據(jù)集數(shù)據(jù)管道和流式處理內(nèi)存受限環(huán)境需要延遲計(jì)算的場(chǎng)景不推薦使用的情況需要多次遍歷數(shù)據(jù)需要隨機(jī)訪問(wèn)元素?cái)?shù)據(jù)集很小且需要頻繁訪問(wèn)7.3 與其他語(yǔ)言的對(duì)比JavaScript的function*和yieldfunction* gen() { yield 1; yield 2; }C#的IEnumerable和yield returnIEnumerableint Gen() { yield return 1; yield return 2; }Ruby的Enumeratorgen Enumerator.new do |y| y 1 y 2 endPython生成器的獨(dú)特優(yōu)勢(shì)在于其簡(jiǎn)潔的語(yǔ)法和與迭代器協(xié)議的無(wú)縫集成。 Python生成器原理與應(yīng)用：從yield到內(nèi)存優(yōu)化
1. 從函數(shù)到生成器的跨越第一次遇到y(tǒng)ield關(guān)鍵字時(shí)我正嘗試處理一個(gè)超過(guò)10GB的日志文件。傳統(tǒng)方法是將整個(gè)文件讀入內(nèi)存結(jié)果自然是內(nèi)存溢出。同事建議我試試生成器從此打開了新世界的大門。yield是Python中一個(gè)神奇的關(guān)鍵字它能讓普通函數(shù)搖身一變成為生成器函數(shù)。與return不同yield會(huì)暫停函數(shù)執(zhí)行并記住當(dāng)前位置的狀態(tài)下次調(diào)用時(shí)從斷點(diǎn)繼續(xù)執(zhí)行。這種特性在內(nèi)存敏感型應(yīng)用中大放異彩。def read_large_file(file_path): with open(file_path) as f: while True: line f.readline() if not line: break yield line這個(gè)簡(jiǎn)單的生成器函數(shù)完美解決了我的內(nèi)存問(wèn)題。它每次只讀取一行到內(nèi)存通過(guò)yield逐行返回而不是一次性加載整個(gè)文件。這就是生成器的核心價(jià)值——按需生成數(shù)據(jù)避免不必要的內(nèi)存消耗。2. 生成器的工作原理剖析2.1 生成器函數(shù)的執(zhí)行流程當(dāng)Python解釋器遇到包含yield的函數(shù)時(shí)不會(huì)像普通函數(shù)那樣立即執(zhí)行而是返回一個(gè)生成器對(duì)象。這個(gè)對(duì)象實(shí)現(xiàn)了迭代器協(xié)議包含以下幾個(gè)關(guān)鍵狀態(tài)創(chuàng)建階段調(diào)用生成器函數(shù)返回生成器對(duì)象此時(shí)代碼尚未執(zhí)行預(yù)激階段首次調(diào)用next()時(shí)代碼執(zhí)行到第一個(gè)yield處暫停掛起階段yield返回右側(cè)表達(dá)式結(jié)果保存所有局部變量狀態(tài)恢復(fù)階段再次調(diào)用next()時(shí)從上次暫停處繼續(xù)執(zhí)行def countdown(n): print(Starting countdown!) while n 0: yield n n - 1 print(Blastoff!) # 創(chuàng)建生成器 counter countdown(3) print(next(counter)) # 輸出: Starting countdown! 然后 3 print(next(counter)) # 輸出 2 print(next(counter)) # 輸出 1 print(next(counter)) # 輸出 Blastoff! 然后拋出StopIteration2.2 生成器的內(nèi)存模型生成器最顯著的優(yōu)勢(shì)是內(nèi)存效率。對(duì)比以下兩種實(shí)現(xiàn)# 傳統(tǒng)列表方式 def squares_list(n): result [] for i in range(n): result.append(i*i) return result # 生成器方式 def squares_gen(n): for i in range(n): yield i*i當(dāng)n1,000,000時(shí)列表版本需要存儲(chǔ)所有計(jì)算結(jié)果內(nèi)存占用約8MB假設(shè)每個(gè)整數(shù)8字節(jié)。而生成器版本在任何時(shí)候只維護(hù)當(dāng)前迭代狀態(tài)內(nèi)存占用幾乎可以忽略不計(jì)。3. yield的高級(jí)用法3.1 生成器表達(dá)式Python提供了更簡(jiǎn)潔的生成器表達(dá)式語(yǔ)法類似于列表推導(dǎo)式# 列表推導(dǎo)式 squares_list [x*x for x in range(10)] # 生成器表達(dá)式 squares_gen (x*x for x in range(10))生成器表達(dá)式特別適合處理大數(shù)據(jù)流比如統(tǒng)計(jì)大型文件中滿足條件的行數(shù)matched_lines sum(1 for line in open(huge.log) if error in line)3.2 yield from語(yǔ)法Python 3.3引入的yield from語(yǔ)法進(jìn)一步簡(jiǎn)化了生成器的嵌套使用def chain(*iterables): for it in iterables: yield from it # 等價(jià)于 def chain_manual(*iterables): for it in iterables: for item in it: yield itemyield from不僅能簡(jiǎn)化代碼還能保持子生成器的返回值def subgenerator(): yield 1 yield 2 return Done def delegator(): result yield from subgenerator() print(fSubgenerator returned: {result}) list(delegator()) # 輸出: Subgenerator returned: Done3.3 協(xié)程與雙向通信生成器通過(guò)send()方法實(shí)現(xiàn)了雙向通信這是協(xié)程的基礎(chǔ)def coroutine(): print(Starting coroutine) while True: received yield print(fReceived: {received}) c coroutine() next(c) # 預(yù)激生成器 c.send(Hello) # 輸出: Received: Hello c.send(World) # 輸出: Received: World這種模式在異步編程中非常有用雖然現(xiàn)代Python更推薦使用async/await語(yǔ)法但理解其底層原理很有必要。4. 實(shí)戰(zhàn)中的陷阱與技巧4.1 生成器只能消費(fèi)一次新手常犯的錯(cuò)誤是重復(fù)使用已耗盡的生成器gen (x for x in range(3)) print(list(gen)) # [0, 1, 2] print(list(gen)) # [] 第二次為空解決方案是重新創(chuàng)建生成器或者使用itertools.tee進(jìn)行復(fù)制注意內(nèi)存開銷。4.2 預(yù)激生成器的必要性需要接收數(shù)據(jù)的生成器必須先調(diào)用next()或send(None)進(jìn)行預(yù)激def echo(): while True: received yield print(received) e echo() e.send(hello) # 報(bào)錯(cuò): cant send non-None value to a just-started generator next(e) # 預(yù)激 e.send(hello) # 正常輸出: hello4.3 性能優(yōu)化技巧雖然生成器節(jié)省內(nèi)存但調(diào)用開銷比列表迭代大。對(duì)于小數(shù)據(jù)集直接使用列表可能更快# 測(cè)試代碼 import timeit small_data range(100) large_data range(1000000) def test_list(data): return [x*x for x in data] def test_gen(data): return list(x*x for x in data) # 小數(shù)據(jù)測(cè)試 print(timeit.timeit(lambda: test_list(small_data), number10000)) # 約0.3秒 print(timeit.timeit(lambda: test_gen(small_data), number10000)) # 約0.4秒 # 大數(shù)據(jù)測(cè)試 print(timeit.timeit(lambda: test_list(large_data), number1)) # 約0.5秒高內(nèi)存 print(timeit.timeit(lambda: test_gen(large_data), number1)) # 約0.6秒低內(nèi)存4.4 調(diào)試生成器調(diào)試生成器可能比較棘手因?yàn)閳?zhí)行流程不是線性的。我常用的方法是添加打印語(yǔ)句def debug_gen(): for i in range(3): print(fYielding {i}) yield i print(fResumed after {i})使用Python 3.7的breakpoint()def debug_gen(): for i in range(3): breakpoint() # 進(jìn)入pdb調(diào)試器 yield i將生成器轉(zhuǎn)換為列表查看所有值注意內(nèi)存消耗gen some_generator() print(list(gen)) # 查看所有輸出5. 生成器在標(biāo)準(zhǔn)庫(kù)中的應(yīng)用Python標(biāo)準(zhǔn)庫(kù)中大量使用了生成器模式典型例子包括5.1 itertools模塊itertools提供了豐富的生成器工具import itertools # 無(wú)限計(jì)數(shù)器 counter itertools.count(start10, step2) print(next(counter)) # 10 print(next(counter)) # 12 # 排列組合 perms itertools.permutations(ABC, 2) print(list(perms)) # [(A, B), (A, C), (B, A), ...] # 分組操作 groups itertools.groupby(AAABBBCCAAA) print([(k, list(g)) for k, g in groups]) # [(A, [A, A, A]), ...]5.2 上下文管理器contextlib.contextmanager裝飾器可以用生成器實(shí)現(xiàn)上下文管理器from contextlib import contextmanager contextmanager def timed_block(label): start time.time() try: yield finally: end time.time() print(f{label} took {end-start:.2f} seconds) with timed_block(calculation): time.sleep(1) # 模擬耗時(shí)操作5.3 文件處理csv模塊的reader函數(shù)返回生成器避免一次性加載大文件import csv def process_large_csv(filepath): with open(filepath) as f: reader csv.reader(f) for row in reader: yield process_row(row) # 逐行處理6. 生成器與異步編程雖然現(xiàn)代Python使用async/await語(yǔ)法處理協(xié)程但理解其與生成器的關(guān)系很有幫助6.1 歷史演變Python異步編程經(jīng)歷了多個(gè)階段生成器yield/sendPython 2.5asyncio.coroutineyield fromPython 3.4async/awaitPython 3.56.2 底層相似性async/await本質(zhì)上是生成器語(yǔ)法糖# 傳統(tǒng)生成器協(xié)程 def old_coroutine(): yield from asyncio.sleep(1) return 42 # 現(xiàn)代async協(xié)程 async def new_coroutine(): await asyncio.sleep(1) return 426.3 實(shí)際應(yīng)用案例生成器非常適合實(shí)現(xiàn)簡(jiǎn)單的狀態(tài)機(jī)def traffic_light(): while True: yield red yield green yield yellow light traffic_light() print(next(light)) # red print(next(light)) # green print(next(light)) # yellow另一個(gè)實(shí)用案例是分塊處理數(shù)據(jù)def chunker(iterable, size): for i in range(0, len(iterable), size): yield iterable[i:isize] for chunk in chunker(range(100), 10): process(chunk) # 每次處理10個(gè)元素7. 性能對(duì)比與最佳實(shí)踐7.1 生成器vs列表的內(nèi)存對(duì)比通過(guò)memory_profiler實(shí)測(cè)內(nèi)存使用profile def list_version(): return [i*i for i in range(1000000)] profile def gen_version(): return (i*i for i in range(1000000)) list_version() # 內(nèi)存峰值約40MB gen_version() # 內(nèi)存峰值基本不變7.2 何時(shí)使用生成器推薦使用生成器的場(chǎng)景處理大型或無(wú)限數(shù)據(jù)集數(shù)據(jù)管道和流式處理內(nèi)存受限環(huán)境需要延遲計(jì)算的場(chǎng)景不推薦使用的情況需要多次遍歷數(shù)據(jù)需要隨機(jī)訪問(wèn)元素?cái)?shù)據(jù)集很小且需要頻繁訪問(wèn)7.3 與其他語(yǔ)言的對(duì)比JavaScript的function*和yieldfunction* gen() { yield 1; yield 2; }C#的IEnumerable和yield returnIEnumerableint Gen() { yield return 1; yield return 2; }Ruby的Enumeratorgen Enumerator.new do |y| y 1 y 2 endPython生成器的獨(dú)特優(yōu)勢(shì)在于其簡(jiǎn)潔的語(yǔ)法和與迭代器協(xié)議的無(wú)縫集成。