Python生成器：我处理10万条数据，内存只占了1MB

2026-02-28

大家好，我是正在实战各种AI项目的程序员晚枫。

今天分享一个让我处理大数据时内存占用减少90%的技术——生成器（Generator）。

你可能遇到过这种情况：要处理几万、几十万条数据，程序直接卡死或报MemoryError。这时候生成器就是你的救星。

看完这篇文章，你会理解为什么生成器被称为”省内存的神器”。

问题：列表太占内存

假设你要处理100万个数字：

# 创建一个包含100万个数字的列表
def get_numbers(n):
    result = []
    for i in range(n):
        result.append(i * i)
    return result

numbers = get_numbers(1000000)
# 这个列表占用约38MB内存！

for num in numbers:
    print(num)  # 处理每个数字

问题：所有数据一次性加载到内存，数据量大了就崩溃。

解决方案：生成器

什么是生成器？

生成器是一种特殊的迭代器，它按需生成数据，而不是一次性全部生成。

就像自动售货机，你要一个它给一个，而不是先把所有商品堆在你面前。

创建生成器的两种方式

方式1：生成器函数（yield）

def get_numbers_generator(n):
    for i in range(n):
        yield i * i  # 用yield代替return

numbers = get_numbers_generator(1000000)
# 几乎不占内存！

for num in numbers:
    print(num)  # 每次只生成一个数字

方式2：生成器表达式

# 列表推导式（占内存）
squares_list = [i * i for i in range(1000000)]

# 生成器表达式（省内存）
squares_gen = (i * i for i in range(1000000))

yield的工作原理

def simple_generator():
    print("开始")
    yield 1
    print("继续")
    yield 2
    print("结束")
    yield 3

gen = simple_generator()
print(next(gen))  # 输出：开始 \n 1
print(next(gen))  # 输出：继续 \n 2
print(next(gen))  # 输出：结束 \n 3
# print(next(gen))  # StopIteration异常

关键点：

遇到yield暂停，保存当前状态
next()被调用时，从暂停处继续执行
函数执行完毕抛出StopIteration

实战案例

案例1：读取大文件

# 传统方式（可能内存爆炸）
def read_lines_bad(filename):
    with open(filename, 'r') as f:
        return f.readlines()  # 一次性读入所有行

# 生成器方式（一行一行读）
def read_lines_good(filename):
    with open(filename, 'r') as f:
        for line in f:
            yield line.strip()

# 使用
for line in read_lines_good('huge_file.txt'):
    process(line)  # 每次只处理一行

案例2：无限序列

def fibonacci():
    """生成无限的斐波那契数列"""
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# 使用
fib = fibonacci()
for _ in range(10):
    print(next(fib))  # 0, 1, 1, 2, 3, 5, 8, 13, 21, 34

案例3：数据管道

def read_log_file(filename):
    """读取日志文件"""
    with open(filename, 'r') as f:
        for line in f:
            yield line.strip()

def filter_errors(lines):
    """过滤错误日志"""
    for line in lines:
        if 'ERROR' in line:
            yield line

def parse_timestamp(lines):
    """解析时间戳"""
    for line in lines:
        # 假设格式：[2024-01-01 10:00:00] ERROR ...
        timestamp = line[1:20]
        yield {'time': timestamp, 'message': line}

# 组合成管道
logs = read_log_file('app.log')
errors = filter_errors(logs)
parsed = parse_timestamp(errors)

for error in parsed:
    print(error)
    # 全程内存占用极小！

生成器 vs 迭代器

特性	生成器	迭代器
创建方式	yield或生成器表达式	__iter__和__next__方法
代码复杂度	简单	较复杂
内存占用	极低	低
可重用性	一次性	一次性

结论：能用生成器就用生成器，代码简洁又高效。

常用内置生成器函数

# enumerate - 带索引的遍历
for i, value in enumerate(['a', 'b', 'c']):
    print(i, value)  # 0 a, 1 b, 2 c

# zip - 并行遍历多个序列
names = ['Alice', 'Bob']
ages = [25, 30]
for name, age in zip(names, ages):
    print(name, age)  # Alice 25, Bob 30

# map - 对每个元素应用函数
squares = map(lambda x: x**2, [1, 2, 3, 4])
print(list(squares))  # [1, 4, 9, 16]

# filter - 过滤元素
evens = filter(lambda x: x % 2 == 0, [1, 2, 3, 4])
print(list(evens))  # [2, 4]

# range - 本身就是生成器
for i in range(1000000):  # 不占用内存
    pass