☰
列表推导式、生成器表达式与生成器函数:内存与选型指南
2026/10/31 14:04:30 网站建设 项目流程

1. 三个概念到底在解决什么问题

很多人第一次接触列表推导式、生成器表达式、生成器函数,都是在某篇教程里看到一行紧凑的写法,觉得"哇,还能这么写",然后就把它当成了炫技工具到处套。结果项目一上线,内存曲线像坐了过山车,日志里冒出一串MemoryError或者更难查的"第二次遍历数据全空了"。这三个东西本质上不是语法糖问题,而是数据生产与消费模式的问题,选错了模型,代码写得多漂亮都会翻车。

我先把话说在前面:这三者的核心差异只有一句话——列表推导式是一次性把结果全部算出来放进内存;生成器表达式和生成器函数是算出下一个就交出去,算过的就不再保留。前者像是把整条流水线的产品全堆进仓库再统一发货,后者是即生产即装车。仓库大就无所谓,仓库小就要出事。这篇文章适合两类人看:一类是刚学 Python、会用推导式但说不清什么时候不该用的人;另一类是写了几年代码、被生成器的"只能消费一次"坑过、想彻底搞明白yield执行流程的人。下面我按"为什么这么设计、怎么用不出错、出错怎么查、项目里怎么定规范"的顺序讲,尽量每个结论都带上能直接跑的代码。

2. 先建立一张选择地图再动手

2.1 用流水线仓库的比喻理解三者定位

把数据处理想成一个车间:原始物料是[1, 2, 3, ..., 1000000],加工工序是"平方后再筛选出偶数"。列表推导式[x*x for x in data if (x*x) % 2 == 0]的做法是——把所有物料一次性搬进车间,全部加工完,成品整整齐齐码进一个新仓库,然后才交货。这个仓库有多大?如果成品有 100 万个整数,光是指针数组就要 8MB 左右(64 位环境下每个指针 8 字节),再加上每个整数对象本身的开销(CPython 里一个普通 int 对象约 28 字节),实际占用很容易冲到 30MB 以上。看起来还能接受?那换成 1000 万行日志切分出来的字典记录呢?每个元素是个 dict,几百字节起步,直接就是几个 GB。这就是列表推导式在数据量上去之后最先暴露的问题:它的内存占用和结果集大小成正比,和你需要多少条数据无关。

生成器表达式(x*x for x in data if (x*x) % 2 == 0)的做法是——车间门口挂了个取号机,你喊一声"下一个",它才转身去拿一份物料加工一份交给你,交完就忘,不留库存。整个生成器对象本身在内存里只占两三百字节量级,剩下的开销全在数据消费端。生成器函数的思路完全一样,只是把加工逻辑从"一个表达式"扩展成了"一个带yield的函数体",可以有循环、有分支、有局部变量、有 try/finally,工序再复杂也能装得下。

2.2 三种写法在同一份数据上的行为对照

光讲概念容易飘,直接上可对照的代码。假设我们要统计一份大文件里每行的长度,只保留长度超过阈值的行。

# 方案 A:列表推导式 lines = [line.strip() for line in open("access.log", encoding="utf-8")] long_lines = [len(line) for line in lines if len(line) > 80]

这段代码有两个问题。第一,open返回的文件对象被推导式隐式遍历,但文件句柄没有显式关闭,虽然在 CPython 里靠引用计数能回收,语义上依然是坏习惯。第二,lines和long_lines两份完整列表同时驻留内存,文件小无所谓,文件一旦到几个 GB 就是灾难。

# 方案 B:生成器表达式 + 链式管道 with open("access.log", encoding="utf-8") as f: lengths = (len(line.strip()) for line in f) long_lengths = (n for n in lengths if n > 80) total = sum(long_lengths)

这份代码在任何文件大小下内存都是恒定的,因为f是逐行读的,lengths和long_lengths都是生成器,来一行处理一行。这里有个关键细节:with块里定义生成器,但生成器真正的迭代发生在sum(long_lengths)那一行,此时文件还开着,所以没问题。如果把long_lengths从with块里返回出去、在外面再消费,文件已经关闭,就会抛ValueError: I/O operation on closed file。这是新手最常踩的坑之一,后面第 5 章会专门列。

# 方案 C:生成器函数,适合逻辑复杂的场景 def iter_long_line_lengths(path, threshold=80): with open(path, encoding="utf-8") as f: for raw in f: n = len(raw.strip()) if n > threshold: yield n total = sum(iter_long_line_lengths("access.log"))

方案 C 把文件生命周期管理封进了函数内部,调用方拿到的只是一个可迭代对象,不用关心文件什么时候开、什么时候关,也不会出现"生成器还没消费完文件就被关了"的问题。这就是我在项目里最推荐的形式:只要处理逻辑超过一层表达式能说清的程度,就写成生成器函数,不要硬堆生成器表达式。

2.3 选择判据不是"数据量大小"而是"消费次数"

很多人以为选型标准是"数据多就用生成器",其实更准确的判据是这份结果你要消费几次。如果结果要遍历两遍、要随机访问、要排序、要传给需要len()的函数,那必须是列表——生成器根本不支持下标、不支持len()、不支持第二次遍历。反过来,如果结果只是喂给sum()、max()、any()、all()、"".join()这类"一遍扫完就结束"的消费者,生成器就是天然最优解。

消费方式列表推导式生成器表达式/函数
单次聚合(sum/max/any/all)可用但浪费内存首选
需要多次遍历必须不可行
需要下标访问、切片必须不可行
需要len()必须不可行(要先转 list)
需要排序必须(或sorted(gen)返回列表)只能作为sorted的入参
数据量超大/流式来源内存风险高首选
管道式多级加工每级都占一份内存首选

这张表我建议直接贴到自己项目的代码规范文档里,比写一堆说明管用。

3. 列表推导式:写得短不等于写得对

3.1 语法结构与真实执行顺序拆解

列表推导式的完整形态是[表达式 for 变量 in 可迭代对象 if 条件],顺序不能颠倒,if也必须在for之后。很多人写惯了 SQL 的SELECT ... WHERE ... FROM ...,第一次写推导式会把if放到前面,直接语法错误。理解执行顺序的方法很简单:把它当成嵌套循环来读,从左往右就是从外到内。

# 单层 squares = [x * x for x in range(10) if x % 2 == 0] # 等价于 squares = [] for x in range(10): if x % 2 == 0: squares.append(x * x)

多层的情况是踩坑重灾区。[x + y for x in xs for y in ys]的顺序是x在外层、y在内层,等价于两层嵌套循环,先固定x再遍历所有y。我见过不少人把矩阵转置写成了[row[i] for row in matrix for i in range(len(matrix[0]))],结果拿到一个扁平列表而不是嵌套结构,排查半天。转置的正确写法是外层表达式负责构造新行,内层生成一行的元素:

matrix = [[1, 2, 3], [4, 5, 6]] transposed = [[row[i] for row in matrix] for i in range(3)] # [[1, 4], [2, 5], [3, 6]]

注意这里是"先写内层[row[i] for row in matrix],再把它当成外层推导式的表达式"。想不清楚的时候,我有个笨办法但极其有效:先把嵌套循环按原样写出来,然后从最内层开始往外一层层压缩成推导式,压缩完对照一眼缩进层级,基本不会错。

3.2 独立作用域这件事,Python 3 里终于干净了

[x for x in range(3)]执行完之后,x在 Python 3 里是不存在的,抛NameError。这一点和普通for循环完全相反(普通循环的变量会泄漏到外层作用域)。Python 2 时代推导式会泄漏变量,迁移项目时如果代码里有依赖泄漏变量的地方,2 到 3 升级会直接炸。现在不用操心这个问题了,但反过来要注意:推导式内部看不到外层同名变量的中间状态,因为它的迭代变量是独立的局部名字。

x = 100 result = [x for x in range(3)] print(x) # 100,外层 x 没被污染 print(result) # [0, 1, 2]

这个特性带来一个好处:推导式可以安全地在函数里使用,不会因为命名冲突改掉外部状态。但也带来一个隐性的问题:如果你在推导式的表达式部分引用了外部变量,那个引用是求值时实时读取的,不是定义时快照。这在闭包场景里会出问题,第 4 章讲延迟绑定时会细说。

3.3 性能优势是真的,但别为了省那 20% 牺牲可读性

列表推导式比for循环加append快,这个结论没什么争议,原因是推导式在字节码层面省掉了每次循环的方法查找和函数调用开销。我实测过的量级大致是:简单算术加工,100 万次迭代,推导式比append写法快 20% 到 40%,具体倍数跟 Python 版本、加工复杂度都有关系,不用记死数字,知道"能快一截"就够了。

但这里我要泼一盆冷水:性能优势只在循环体足够简单时才明显。如果表达式里每次都要调用一个开销很大的自定义函数,那瓶颈在函数本身,推导式省下的那点字节码开销占比很小。更要命的是可读性——我见过有人把一个带三个条件分支、两个函数调用的加工逻辑硬塞进一行推导式,长度超过 200 个字符,换行折了四次。这种代码 review 的时候没人愿意读,改一个条件要花五分钟定位。我的经验线是:推导式的表达式部分不超过一个函数调用加一个运算符组合,就继续用;超过了就老老实实写生成器函数。

3.4 用推导式做纯副作用操作是错误的用法

# 反例:只为了打印,结果列表完全没用 [print(x) for x in range(5)]

这段代码能跑,输出也对,但它构造了一个全是None的列表然后立刻丢弃,是纯粹的浪费。更严重的问题是,读代码的人会以为这个列表后面还要用,review 时多花时间去确认。推导式的语义是"构造一个集合",不是"执行一串动作"。需要副作用就用普通for循环,一句话的事。

顺带说一句字典和集合推导式。{k: v for k, v in pairs}和{x for x in xs}语法结构和列表推导式一致,只是外层括号不同。字典推导式有个细节:重复键后面的会覆盖前面的,这在做"后写优先"的合并时很方便,但如果本意是去重统计,就得提前想清楚覆盖是不是你要的行为。

4. 生成器表达式:把内存占用按需分配

4.1 惰性求值到底省在哪里

生成器表达式(expr for item in iterable if cond)只做一件事:创建一个生成器对象,把迭代逻辑挂上去,然后立刻返回,一行数据都不算。这个"立刻返回"是理解所有后续行为的关键。你写g = (x*x for x in range(10**8)),这行代码执行时间是微秒级,内存占用是常数级,因为range对象本身是惰性的,生成器又只是包了一层。

真正的计算发生在每次next(g)时。next触发后,生成器内部推进到下一个满足条件的元素,把表达式算出来返回,然后暂停在原处,保存当前状态(循环变量值、迭代器位置、局部变量)。下一次next从暂停点继续。这个"暂停-恢复"机制是生成器区别于普通函数的核心,也是它能有常数内存的根本原因。

对比一下内存量级,感受会更直观:

import sys g = (x for x in range(1_000_000)) print(sys.getsizeof(g)) # 百字节量级 lst = [x for x in range(1_000_000)] print(sys.getsizeof(lst)) # 800 万字节量级(仅指针数组)

sys.getsizeof(lst)只统计列表对象本身,约 8MB,还不包含那一百万个 int 对象。而生成器对象就是几百字节,无论后面挂着多大的数据源。这个差距在小数据上没感觉,在流式处理场景里就是"能跑"和"跑不动"的区别。

4.2 只能消费一次,这是设计不是 bug

生成器最让人崩溃的特性是一次性。它不像列表那样保存所有元素,它只保存"我算到哪了"这个游标。遍历完之后游标到底,再遍历就是空的。

g = (x for x in range(3)) print(list(g)) # [0, 1, 2] print(list(g)) # [] print(sum(g)) # 0

我第一次遇到这个现象时以为是 bug,查了半天文档才明白这是设计使然。所以工程上有一条硬规矩:生成器对象不要在函数之间随意传递后还指望别人能复用。如果确实需要多次使用,要么在创建处就转成列表(data = list(gen)),要么每次用的时候重新调用生成器函数生成一个新的。后者通常是更好的选择,因为转列表就把内存优势丢掉了。

这里有个隐蔽的变体:zip和生成器一起用的时候,如果其中一个参数被提前消费过,结果会静默变短,不报错。这种 bug 最难查,因为程序不崩,只是数据少了一截。排查办法是在可疑位置插一句print(len(list(...))),但注意list()本身又消费了一次,所以只能在调试副本上做,不能改原逻辑。

4.3 函数调用里可以省掉一层括号,但别滥用

生成器表达式作为唯一参数传给函数时,外层括号可以省略:

total = sum(x * x for x in range(100)) joined = ",".join(str(x) for x in range(10)) any_even = any(x % 2 == 0 for x in range(10))

这种写法在sum、any、all、join、max、min上非常常见,读起来也自然。但如果有第二个参数,括号就不能省:

total = sum((x * x for x in range(100)), 0) # 有初始值,必须带括号

注意上面这个坑:sum(gen, 0)不写内层括号会变成sum收到两个参数,语法上其实sum(x*x for x in range(100), 0)是非法语法,直接报错,不会静默出错,还算友好。

还有个更值得警惕的:any/all遇到生成器会短路,这是好事,但意味着后面可能有副作用没执行。any(process(x) for x in items)在遇到第一个True时立刻停手,process只被调用了部分次数。如果你的process有写日志、改状态之类的副作用,行为就和"全部处理一遍"完全不同。这不是 bug,但必须显式知道。

4.4 生成器表达式里的延迟绑定陷阱

这是个经典问题,值得单独拎出来。生成器的循环变量在表达式部分被引用时,是延迟绑定的:闭包捕获的是变量名,不是当时的值。

funcs = [lambda: x for x in range(3)] print([f() for f in funcs]) # [2, 2, 2],不是 [0, 1, 2]

三个 lambda 都捕获了同一个变量x,等它们真正被调用时,x已经是循环结束后的值 2。修法是用默认参数在定义时快照:[lambda x=x: x for x in range(3)]。同样的问题在map和生成器表达式里都会出现,尤其在构造"回调列表"时特别容易中招。我的建议是:只要推导式或生成器表达式里出现了lambda或定义了内部函数,就停下来检查绑定时机,这一步花十秒,能省下两小时排查。

4.5 和列表推导式的性能对照,别迷信"生成器一定更快"

这里要纠正一个常见误解。生成器省内存是确定的,但单次迭代速度通常略慢于列表推导式,因为每次next都有状态切换开销。我做过一组粗略对比,处理 100 万条简单整数运算:列表推导式总耗时更短,但峰值内存高出一个数量级;生成器表达式总耗时略长,但峰值内存几乎不变。

所以结论是分场景的:如果结果集不大(比如几万条以内)、并且要多次使用,用列表推导式;如果数据量大或者来源是流式的(文件、网络、数据库游标),用生成器,牺牲一点速度换内存。千万别为了"看起来更高级"把所有推导式都改成生成器,然后在需要len()的地方到处套list(),那才是真正的不伦不类——既没省内存,又多了两层转换。

5. 生成器函数:把 yield 的执行流程彻底搞明白

5.1 yield 不是 return,它是"交出去然后原地待命"

理解yield最快的方式是把它和return对照:return是"我给结果,函数结束,栈帧销毁";yield是"我先给你一个结果,函数暂停,栈帧保留,等你下次来叫我再继续"。

def counter(n): print("开始") for i in range(n): print(f"准备交出 {i}") yield i print(f"从 {i} 之后恢复") print("结束") return "done" g = counter(2) print(next(g)) # 开始 / 准备交出 0 / 0 print(next(g)) # 从 0 之后恢复 / 准备交出 1 / 1 print(next(g)) # 从 1 之后恢复 / 结束 / 抛 StopIteration

把这段代码跑一遍,打印顺序会让你对"暂停点在哪"有非常直观的认识。几个关键点:第一次调用counter(2)什么都不会执行,连"开始"都不打印,只是创建了生成器对象;第一次next才真正开始跑,执行到第一个yield暂停;每次恢复都从yield语句的下一行继续;函数体跑完时抛StopIteration,for循环会自动捕获它并正常结束。

生成器对象有四种状态,调试时用inspect.getgeneratorstate()能直接看到:GEN_CREATED(刚创建未启动)、GEN_RUNNING(正在执行)、GEN_SUSPENDED(停在 yield 处)、GEN_CLOSED(已结束)。排查"为什么这个生成器没输出"的时候,先看一眼状态,比加一堆 print 快得多。

5.2 send、throw、close:生成器的三个控制手柄

send是进阶用法,它能在恢复生成器的同时把一个值送进去,yield表达式本身会接收到这个值。

def averager(): total = 0.0 count = 0 avg = None while True: value = yield avg if value is None: break total += value count += 1 avg = total / count return avg g = averager() next(g) # 必须先预热到第一个 yield,返回 None print(g.send(10)) # 10.0 print(g.send(20)) # 15.0 print(g.send(30)) # 20.0

这里有个必须记住的规矩:刚创建的生成器必须先next()或send(None)预热一次,让它跑到第一个yield处挂起,之后才能send非 None 的值。直接g.send(10)会抛TypeError: can't send non-None value to a just-started generator。这个设计的原因是第一次执行时还没有任何yield表达式在等待接收值,无处可送。

throw往生成器内部注入一个异常,在yield处抛出,可以用来做错误通知。close则是通知生成器"我不要了",它在挂起处抛GeneratorExit,生成器应该借这个机会做清理(关文件、释放连接),然后就结束了。如果你的生成器里有try/finally,close会触发finally块,这是资源清理的正确姿势:

def read_lines(path): f = open(path, encoding="utf-8") try: for line in f: yield line.rstrip("\n") finally: f.close() print("文件已关闭") g = read_lines("data.txt") next(g) g.close() # 打印"文件已关闭"

如果生成器在GeneratorExit之后还试图yield,Python 会抛RuntimeError: generator ignored GeneratorExit。所以清理逻辑里绝对不能有yield,这是明确的红线。

5.3 yield from:委托给子生成器,顺便拿到返回值

yield from的作用是把迭代工作委托给另一个可迭代对象,同时把子生成器的return值作为整个yield from表达式的值返回。这在"扁平化嵌套结构"和"组装处理管道"时特别好用。

def flatten(items): for item in items: if isinstance(item, list): yield from flatten(item) else: yield item print(list(flatten([1, [2, [3, 4]], 5]))) # [1, 2, 3, 4, 5]

递归扁平化是yield from最经典的用法,比自己写栈维护可读性好太多。另一个用法是提取子生成器的返回值:

def sub(): yield 1 yield 2 return "子生成器完成" def outer(): result = yield from sub() print(f"收到:{result}") yield 3 print(list(outer())) # 收到:子生成器完成 / [1, 2, 3]

这个机制在写复杂的数据管道时很有价值:每一级处理完可以返回一个统计信息(处理了多少条、跳过了多少条),上层统一收集。我在做日志清洗管道时就靠这个把"处理条数"和"跳过条数"一路带上来,不用额外维护全局变量。

5.4 生成器函数在工程里的四类典型场景

第一类是流式读取大文件,前面已经演示过。第二类是无穷序列,比如自增 ID、斐波那契数列、按需生成的时间片,生成器天生适合表达"无限但按需取"的结构,配合itertools.islice就能安全截断:

import itertools def fib(): a, b = 0, 1 while True: yield a a, b = b, a + b print(list(itertools.islice(fib(), 10))) # 前 10 个

第三类是多级管道加工,每一级都是一个生成器函数,串起来形成 lazy pipeline,任意一级都可以独立测试。第四类是状态机与任务调度,yield天然表达"处理到一半先让出去,下次接着干",早期很多协程实现就是基于这个特性。不过在现在这个年代,真正的并发调度应该用asyncio的async/await,普通生成器不要硬扛并发任务,容易写出难以调试的代码。

6. 常见问题与排查技巧实录

6.1 高频报错速查表

现象/报错根本原因处理方式
TypeError: 'generator' object is not subscriptable生成器不支持下标转list()或用itertools.islice
第二次遍历结果为空生成器只能消费一次每次重新创建,或提前转列表
sum()返回 0生成器已被消费完检查是否在前面被list()或for走过一遍
TypeError: object of type 'generator' has no len()生成器不预知长度需要长度就只能物化
ValueError: I/O operation on closed file生成器在with块外被消费把消费逻辑放进with,或封装成生成器函数
RuntimeError: generator ignored GeneratorExitfinally里还有yield清理逻辑中禁止yield
TypeError: can't send non-None value to a just-started generator未预热就send先next(g)一次
ValueError: generator already executing多线程/重入同时推进同一个生成器生成器不是线程安全的,加锁或每线程独立生成

这张表里出现频率最高的是前三条,本质都是"一次性"引发的。我的排查习惯是:看到结果莫名其妙变少或变空,第一反应就是找这个对象在哪被消费过。在可疑对象上加一行包装,把消费点打出来,通常五分钟内能定位。

6.2 那个让我加班到晚上的 with 块 bug

讲个具体案例。有个批处理脚本,逻辑是读一个大文件、过滤、写回另一个文件,代码大致是这样:

def build_pipeline(path): with open(path, encoding="utf-8") as f: return (line.strip() for line in f if line.strip()) lines = build_pipeline("big.log") count = sum(1 for _ in lines) # 这里直接报错

报错是ValueError: I/O operation on closed file。原因很直白:build_pipeline返回时with块已经退出,文件关闭了,但生成器还没开始迭代,等外面真正消费时无文件可读。这个 bug 的隐蔽之处在于,如果函数里写的是return [line.strip() for line in f],因为列表推导式在with块内就把数据全算完了,反而不会报错。同一段逻辑,把列表推导式改成生成器表达式,就从"正常"变成"报错",这是重构时最容易忽略的语义变化。

修法有两个:一是把消费逻辑挪进with块,二是干脆把整个函数改成生成器函数,让文件生命周期跟着迭代走(就是第 2 章方案 C 的写法)。我后来统一采用第二种,因为它把资源管理和数据生产绑在了一起,调用方怎么写都不会错。这个教训我写进了团队规范:凡是返回生成器表达式、且内部持有需要关闭的资源,一律改成生成器函数。

6.3 调试生成器的三个实用手法

第一个手法是临时物化观察:在可疑位置把生成器转成list看一眼内容,确认数据对不对。注意这只能用在调试副本上,因为物化本身会消费掉生成器,改回原逻辑时要删掉。

第二个手法是用itertools.tee分叉。如果确实需要在两个地方消费同一份流,tee可以复制出两个独立的迭代器:

import itertools g = (x for x in range(5)) a, b = itertools.tee(g) print(list(a)) # [0, 1, 2, 3, 4] print(list(b)) # [0, 1, 2, 3, 4]

但tee不是免费的:它内部会缓存两个分支进度差之间的所有元素,如果两个分支消费速度差距很大,缓存会一路膨胀,内存优势就没了。所以tee只适合"两个消费者进度接近"的场景,别拿它当万能复制器。

第三个手法是在生成器函数里插日志。生成器函数的执行是分段进行的,普通断点调试有时候不好跟,直接在yield前后打日志反而最直观:

def process(items): for i, item in enumerate(items): result = transform(item) print(f"[debug] 第 {i} 条:{item!r} -> {result!r}") yield result

7. 工程落地:选型策略与团队规范

7.1 按数据规模和消费模式分档决策

我把实际项目里的选择整理成三档,比"看情况"这种模糊说法好用。

小数据(万级以内)、需要多次使用:直接用列表推导式,别想太多。这个量级内存开销可以忽略,可读性和便利性优先。中等数据(十万到百万级)、单次聚合:用生成器表达式,一行搞定,内存立省一个数量级。大数据或流式来源(文件、游标、网络流):用生成器函数,把资源管理和加工逻辑封在一起,绝对不要在中间环节物化成列表。

还有一个经常被忽略的档位:需要排序或分组的中间结果。sorted()和groupby()都需要把数据读进来,这时候生成器的内存优势会在这个点上被打回原形。如果数据量真的很大,正确做法不是硬扛,而是考虑外部排序或者分批处理,生成器只是延迟了内存峰值,不能消除它。

7.2 团队代码规范里该写死的几条

第一条:推导式只允许一层for和一个if,超过就改写生成器函数。这条规矩看起来保守,但能挡掉绝大部分难以 review 的代码。

第二条:函数返回可迭代对象时,类型注解写Iterator[T]而不是list[T],让调用方明确知道这玩意儿只能消费一次。如果返回的是列表,就写list[T]。类型注解在这里是文档,能救命。

from typing import Iterator def read_records(path: str) -> Iterator[dict]: ...

第三条:涉及文件、连接、锁的生成器,必须用try/finally或上下文管理器保证释放,不能指望调用方一定会把生成器消费完。生成器被提前丢弃(比如any()短路)是常态,不清理就会泄漏资源。

第四条:不要跨线程共享生成器对象。生成器内部状态不是线程安全的,两个线程同时next会抛ValueError: generator already executing,更糟的情况是状态错乱但不报错。正确做法是每个线程各自创建生成器。

7.3 一个可以直接抄的管道模板

最后给一个我在多个项目里复用过的结构,把"读、清洗、过滤、聚合"四段拆成独立生成器函数,每段都能单独测试:

import itertools from typing import Iterable, Iterator, Any def read_source(path: str) -> Iterator[str]: with open(path, encoding="utf-8", errors="replace") as f: yield from f def clean(lines: Iterable[str]) -> Iterator[str]: for line in lines: s = line.strip() if s: yield s def filter_keyword(lines: Iterable[str], keyword: str) -> Iterator[str]: for line in lines: if keyword in line: yield line def batch(lines: Iterable[str], size: int = 1000) -> Iterator[list[str]]: it = iter(lines) while True: chunk = list(itertools.islice(it, size)) if not chunk: return yield chunk pipeline = batch(filter_keyword(clean(read_source("big.log")), "ERROR"), 500) for chunk in pipeline: handle(chunk) # 分批处理,内存恒定

这个模板的价值在于每一段职责单一、都可以用假数据单独测,batch那一段还顺手解决了"下游需要批量提交(比如入库、发请求)"的现实需求。写的时候注意batch里用的是itertools.islice而不是切片,因为生成器不支持切片。

我在实际项目里踩过的坑总结成一句话:列表推导式负责"我要这份数据",生成器负责"我要这个流程"。把这两件事分清,选型基本不会错。至于那些"生成器一定比列表快"、"推导式越短越牛"的说法,听听就好,代码是给人读的,跑得快只是及格线。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询