★ FEATURED ARTICLE 1. 从一段看起来能跑但内存炸了的代码说起很多人第一次接触列表推导式都是被它那行云流水的写法吸引的——一行顶五行简洁、优雅、Pythonic。但真正在生产环境里踩过坑的人都知道列表推导式、生成器、迭代器这三样东西表面上是语法糖和进阶特性实际上它们背后对应的是三种完全不同的数据处理哲学一次性全量构建、惰性按需产出、以及协议驱动的统一遍历接口。搞不清楚这三者的边界写出来的代码要么内存爆掉要么逻辑诡异到调试半天找不到原因。我见过太多这样的情况一个处理日志文件的脚本用列表推导式读了几十万行数据本地测试没问题一上服务器直接OOM或者有人写了个生成器函数结果在里面用了return返回值调用方拿到的是个生成器对象而不是结果一脸懵。这些问题的根源都不是语法不熟而是对迭代协议和惰性求值的理解停留在表面。这篇内容就是想把这三块知识彻底串起来讲清楚。不管你是刚学Python不久还是已经写了几年代码但对生成器始终会用但说不清我都会从底层机制、实际选型、性能对比、常见坑位几个角度展开。核心关键词就三个列表推导式、生成器、迭代器。读完你应该能做到看到一段数据处理代码立刻判断该用哪种方式以及为什么。2. 列表推导式的本质它到底比for循环快在哪2.1 列表推导式的执行模型先看一段最基础的对比。假设我们要生成0到9的平方列表# 方式一普通for循环 squares [] for i in range(10): squares.append(i ** 2) # 方式二列表推导式 squares [i ** 2 for i in range(10)]两种写法结果完全一样但执行过程有本质区别。普通for循环每次迭代都要调用一次append方法这是一次方法查找加方法调用的开销。而列表推导式在CPython层面被编译成了一种专门的字节码指令LIST_APPEND它直接操作列表对象跳过了属性查找的环节。你可以用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)实测下来在数据量较大的场景比如百万级列表推导式通常比等价的for循环快20%到40%。这个差距在纯计算密集型的任务里非常可观。但要注意这个优势主要来自字节码层面的优化而不是推导式这个概念本身有什么魔法。2.2 条件过滤与嵌套推导式的正确姿势列表推导式支持条件过滤语法是把if放在for后面# 只保留偶数 evens [i for i in range(20) if i % 2 0] # 等价于 evens [] for i in range(20): if i % 2 0: evens.append(i)这里有个容易搞混的点if放在for前面和后面含义完全不同。放在后面是过滤放在前面是三元表达式# 过滤只保留偶数奇数直接丢弃 [x for x in range(10) if x % 2 0] # [0, 2, 4, 6, 8] # 三元所有元素都保留偶数原样奇数变-1 [x if x % 2 0 else -1 for x in range(10)] # [0, -1, 2, -1, 4, ...]嵌套推导式也要小心顺序。[x y for x in [1,2] for y in [10,20]]等价于两层for循环外层是x内层是y结果是[11, 21, 12, 22]。很多人第一次写嵌套推导式会把顺序搞反导致结果不对。2.3 列表推导式的内存陷阱这是最需要警惕的地方。列表推导式会立即构建完整的列表所有元素一次性放进内存。数据量小的时候无所谓但如果你处理的是文件行、数据库记录、网络响应这类可能很大的数据源列表推导式就是内存杀手。# 危险写法如果文件有1000万行这个列表会吃掉大量内存 lines [line.strip() for line in open(huge_file.txt)] # 安全写法用生成器表达式逐行处理 lines (line.strip() for line in open(huge_file.txt))注意上面第二个写法用的是圆括号而不是方括号这就是生成器表达式。它不会一次性构建列表而是返回一个生成器对象每次迭代才产出一个值。内存占用从O(n)降到O(1)。提示判断该用列表推导式还是生成器表达式核心就一句话——这个结果后续需要被多次遍历吗需要索引访问吗需要知道长度吗如果都不需要优先用生成器。3. 生成器惰性求值背后的状态机机制3.1 生成器函数与yield的执行流程生成器函数就是包含yield关键字的函数。调用它不会执行函数体而是返回一个生成器对象def count_up(n): print(开始执行) i 0 while i n: yield i i 1 print(执行结束) gen count_up(3) print(type(gen)) # class generator注意此时开始执行并没有被打印。只有当你调用next(gen)或者用for循环遍历时函数体才真正开始运行。每次遇到yield函数会暂停并交出当前值同时保存所有局部变量的状态。下次调用next时从暂停处继续。这个暂停-恢复的能力底层是靠栈帧保留实现的。普通函数调用结束后栈帧就销毁了而生成器函数的栈帧会被挂起保留直到生成器被耗尽或显式关闭。这也是为什么生成器能记住i的值。gen count_up(3) print(next(gen)) # 打印开始执行然后输出0 print(next(gen)) # 输出1 print(next(gen)) # 输出2 print(next(gen)) # 打印执行结束然后抛StopIteration3.2 生成器表达式与生成器函数的选型生成器表达式适合简单的、一次性的惰性序列total sum(x ** 2 for x in range(1000000))生成器函数适合逻辑复杂、需要多步处理、或者需要复用的情况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选型原则如果逻辑能用一个表达式写完用生成器表达式如果需要多行、有分支、有循环嵌套用生成器函数。不要为了炫技把复杂逻辑硬塞进一个生成器表达式里可读性会急剧下降。3.3 生成器的send、throw与close生成器不只是产出值它还支持双向通信。send方法可以向生成器内部传值这个值会成为当前yield表达式的返回值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用于在生成器内部抛出异常close用于关闭生成器并触发GeneratorExit。这些特性在协程、状态机、管道处理等场景里非常有用但日常业务代码用得不多。我的建议是先把基础的yield和惰性求值吃透send/throw这些等真正需要时再深入。注意生成器只能被遍历一次。遍历完之后再调用next会直接抛StopIteration。如果你需要多次遍历要么重新创建生成器要么把结果转成列表但要考虑内存。4. 迭代器协议for循环背后到底发生了什么4.1 可迭代对象与迭代器的区别这是面试高频题也是实际写代码时最容易混淆的概念。先给结论可迭代对象Iterable实现了__iter__方法的对象能被iter()调用。迭代器Iterator同时实现了__iter__和__next__方法的对象能被next()调用。列表、元组、字典、字符串都是可迭代对象但它们不是迭代器。对它们调用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循环的本质就是先调用iter()拿到迭代器然后不断调用next()直到捕获StopIteration异常为止。用while循环手动模拟就是nums [1, 2, 3] it iter(nums) while True: try: value next(it) except StopIteration: break print(value)4.2 自定义迭代器的完整实现理解了协议就可以自己写迭代器。下面是一个斐波那契数列迭代器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关键点__iter__返回self说明迭代器本身也是可迭代的。__next__在耗尽时必须抛StopIteration这是协议约定不能返回None或者别的什么来表示结束。4.3 迭代器与生成器的关系生成器本质上是一种语法糖形式的迭代器。任何生成器都自动实现了迭代器协议你不需要手动写__iter__和__next__。反过来迭代器不一定非得是生成器上面那个Fibonacci类就是纯手写的迭代器。实际开发中90%的场景用生成器就够了手写迭代器类通常只在需要维护复杂状态、或者需要提供额外方法比如reset时才用。我个人的经验是能用生成器函数解决的绝不手写迭代器类因为生成器的状态管理由解释器负责不容易出错。对比维度列表推导式生成器表达式生成器函数手写迭代器类内存占用O(n)O(1)O(1)O(1)可重复遍历是否否取决于实现支持索引是否否否代码复杂度低低中高适用场景小数据量、需多次访问简单惰性序列复杂惰性逻辑需要额外状态管理5. 性能实测三种方式在不同数据规模下的表现5.1 内存占用的量化对比光说生成器省内存太抽象我们直接测。用sys.getsizeof看对象本身大小再用tracemalloc看实际内存分配import tracemalloc N 1_000_000 # 列表推导式 tracemalloc.start() lst [i * 2 for i in range(N)] current, peak tracemalloc.get_traced_memory() tracemalloc.stop() print(f列表推导式峰值内存: {peak / 1024 / 1024:.2f} MB) # 生成器表达式 tracemalloc.start() gen (i * 2 for i in range(N)) current, peak tracemalloc.get_traced_memory() tracemalloc.stop() print(f生成器表达式峰值内存: {peak / 1024 / 1024:.2f} MB)在百万级数据下列表推导式的峰值内存通常在30到40MB左右取决于整数对象大小而生成器表达式几乎可以忽略不计通常不到1MB。这个差距在千万级、亿级数据下会变成致命的区别。5.2 执行速度的取舍但生成器不是没有代价的。每次next调用都有额外的开销所以在纯速度上列表推导式通常更快import timeit # 列表推导式求和 t1 timeit.timeit([i * 2 for i in range(100000)], number100) # 生成器表达式求和 t2 timeit.timeit(sum(i * 2 for i in range(100000)), number100)实测结果通常是列表推导式快1.5到2倍。原因很简单列表推导式在C层面批量构建而生成器每次产出一个值都要经过Python层面的状态机调度。所以选型的核心权衡是内存敏感还是速度敏感。如果数据量不大几万条以内列表推导式更快更直接如果数据量大或者数据源是流式的文件、网络、传感器生成器是唯一合理的选择。5.3 一个真实场景的选型推演假设你要处理一个日志文件统计所有ERROR级别的日志条数。文件可能很大但统计结果只是一个整数。错误做法lines [line for line in open(app.log)] # 全量加载内存爆炸 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)第二种写法全程只维护一个计数器内存恒定。而且sum配合生成器表达式是Python里非常经典的惰性统计模式值得记住。再进一步如果还要按错误类型分组统计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循环反而比推导式更清晰因为涉及多个操作。不要为了用推导式而用推导式这是很多人容易走偏的地方。6. 那些年我踩过的生成器与迭代器坑6.1 生成器只能消费一次的隐蔽bug这是最经典的坑。看这段代码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)得到空列表因为生成器已经耗尽了。这个bug在函数间传递生成器时特别隐蔽。比如你把生成器传给一个函数做统计又传给另一个函数做过滤第二个函数拿到的就是空的。解决方案有两个要么在需要多次使用的地方转成列表确认内存可控要么每次重新调用生成器函数创建新的生成器对象。我个人的习惯是生成器对象绝不跨函数边界传递超过一次如果确实需要多处使用在源头就明确是列表还是生成器。6.2 在生成器里使用return的返回值问题生成器函数里的return不会返回值给调用方而是触发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然后期望调用方拿到result那一定会失望。6.3 迭代过程中修改集合的灾难nums [1, 2, 3, 4, 5] for n in nums: if n % 2 0: nums.remove(n) print(nums) # [1, 3, 5]不一定在遍历列表的同时修改它会导致迭代器索引错乱结果不可预测。正确做法是遍历副本或者用推导式重建nums [n for n in nums if n % 2 ! 0]字典和集合同理遍历时不能增删元素否则直接抛RuntimeError: dictionary changed size during iteration。这个坑在写数据清洗逻辑时特别容易踩。6.4 生成器里的异常处理边界生成器内部的异常会在next调用时向外传播。如果你在生成器里捕获了异常但没有正确处理可能导致生成器静默终止def risky_gen(): for i in range(5): try: yield 10 / (i - 2) except ZeroDivisionError: yield 0 # 必须yield一个值否则这个位置就跳过了注意i2时除零如果不yield任何东西生成器会继续下一个循环但调用方感知不到这里出过问题。是否要yield一个占位值取决于业务语义。我的建议是生成器内部的异常要么向上抛要么明确产出一个错误标记值不要静默吞掉。7. 把三者串起来一个完整的数据管道实战7.1 需求描述与整体设计假设我们要处理一个CSV格式的传感器数据文件需求是读取文件、过滤掉无效行、把温度从摄氏度转成华氏度、输出前10条结果。文件可能很大不能全量加载。整体设计思路用生成器函数做文件读取和过滤用生成器表达式做转换最后用itertools.islice取前10条。整条管道全程惰性内存恒定。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 每个环节的选型理由read_valid_rows用生成器函数而不是列表推导式因为文件可能很大而且过滤逻辑涉及异常处理用推导式写会很难看。to_fahrenheit也用生成器函数保持管道的惰性。最后的islice是关键——它只从管道里取10条意味着整个文件只会被读取到满足10条有效数据为止后面的内容根本不会被处理。如果这里用列表推导式整个文件会被完整读取和转换然后才取前10条浪费大量计算和内存。这就是惰性管道的威力计算量与实际消费量成正比而不是与数据源大小成正比。7.3 管道调试的实用技巧惰性管道有个调试难点出错时不知道是哪一环的问题。我的做法是在每个环节临时插入一个探针生成器def probe(rows, label): for row in rows: print(f[{label}] {row}) yield row pipeline to_fahrenheit(probe(read_valid_rows(sensor.csv), raw))这样能看到数据在每个环节的形态。调试完把probe去掉即可。另一个技巧是用itertools.tee把一个生成器分成两个一个用于调试输出一个用于正常流程但要注意tee会缓存数据大流量下慎用。提示惰性管道最适合读取-过滤-转换-聚合这类线性流程。如果流程中有分支、需要回溯、或者需要多次遍历同一批数据就该考虑在某个节点落地成列表或写入临时存储。8. 关于这三块知识我个人的几条经验写了这么多年Python关于列表推导式、生成器、迭代器我总结了几条不太会在教程里看到但很实用的经验。第一条列表推导式不要超过两层嵌套。三层以上的嵌套推导式可读性比等价的for循环差得多维护成本极高。我见过有人写出四层嵌套的推导式同事看了十分钟没看懂最后重写成for循环反而一目了然。简洁不等于短可读性永远优先。第二条生成器的价值不只是省内存更是解耦生产与消费。当你把数据生产逻辑写成生成器消费方可以按自己的节奏拉取两者通过迭代协议解耦。这个思路在写爬虫、消息处理、流式计算时特别有用。生产者不需要知道消费者什么时候要数据消费者也不需要知道数据怎么来的。第三条迭代器协议是理解Python很多特性的钥匙。for循环、解包、zip、map、filter、itertools、yield from、异步的async for全都建立在迭代协议之上。把__iter__和__next__这两个方法吃透很多看似无关的知识点会自动串起来。第四条性能优化先测量再动手。不要凭感觉认为生成器一定比列表快或者慢。用timeit和tracemalloc实测用数据说话。我遇到过数据量只有几千条却非要用生成器的情况结果代码复杂度上去了性能反而因为状态机开销略有下降。工具没有绝对的好坏只有适不适合当前场景。最后分享一个我常用的判断口诀要多次用、要索引、数据小用列表只遍历一次、数据大、流式来源用生成器需要自定义复杂状态才手写迭代器类。这三句话覆盖了日常开发中绝大多数选型场景剩下的边角情况等你真正遇到了自然会有判断。
阅读完成 · 觉得有帮助？
