Python 迭代器与生成器完全指南：从原理到实战 | Python 进阶核心知识

1
# 可迭代对象（Iterable）：实现了 __iter__ 方法
2
# 迭代器（Iterator）：实现了 __iter__ 和 __next__ 方法
3

4
# 列表是可迭代对象，但不是迭代器
5
nums = [1, 2, 3]
6
print(hasattr(nums, '__iter__'))  # True
7
print(hasattr(nums, '__next__'))  # False
8

9
# 获取迭代器
10
it = iter(nums)
11
print(hasattr(it, '__iter__'))   # True
12
print(hasattr(it, '__next__'))   # True
13

14
# 使用 next() 获取元素
15
print(next(it))  # 1
16
print(next(it))  # 2
17
print(next(it))  # 3
18
print(next(it))  # StopIteration

1
# 这段 for 循环：
2
for item in [1, 2, 3]:
3
    print(item)
4

5
# 实际上等价于：
6
it = iter([1, 2, 3])
7
while True:
8
    try:
9
        item = next(it)
10
        print(item)
11
    except StopIteration:
12
        break

1
class Countdown:
2
    """从 N 倒数到 1 的迭代器"""
3

4
    def __init__(self, start):
5
        self.current = start
6

7
    def __iter__(self):
8
        return self  # 返回自身，因为本身就是迭代器
9

10
    def __next__(self):
11
        if self.current <= 0:
12
            raise StopIteration
13

14
        value = self.current
15
        self.current -= 1
16
        return value
17

18
# 使用
19
for i in Countdown(5):
20
    print(i, end=' ')
21
# 输出: 5 4 3 2 1
22

23
# 手动使用
24
cd = Countdown(3)
25
print(next(cd))  # 3
26
print(next(cd))  # 2
27
print(next(cd))  # 1

1
class Range:
2
    """自定义 range 类（可迭代对象，但不是迭代器）"""
3

4
    def __init__(self, start, end, step=1):
5
        self.start = start
6
        self.end = end
7
        self.step = step
8

9
    def __iter__(self):
10
        # 返回一个新的迭代器对象
11
        return RangeIterator(self.start, self.end, self.step)
12

13
class RangeIterator:
14
    """Range 的迭代器"""
15

16
    def __init__(self, start, end, step):
17
        self.current = start
18
        self.end = end
19
        self.step = step
20

21
    def __iter__(self):
22
        return self
23

24
    def __next__(self):
25
        if self.current >= self.end:
26
            raise StopIteration
27
        value = self.current
28
        self.current += self.step
29
        return value
30

31
# 可以多次迭代（因为每次 __iter__ 都返回新迭代器）
32
r = Range(0, 5, 2)
33
for i in r:
34
    print(i, end=' ')  # 0 2 4
35
print()
36
for i in r:
37
    print(i, end=' ')  # 0 2 4（可以再次迭代）

1
def countdown(n):
2
    """倒数生成器"""
3
    while n > 0:
4
        yield n
5
        n -= 1
6

7
# 调用生成器函数返回生成器对象
8
gen = countdown(5)
9
print(type(gen))  # <class 'generator'>
10

11
# 生成器是迭代器
12
print(hasattr(gen, '__iter__'))  # True
13
print(hasattr(gen, '__next__'))  # True
14

15
# 使用 next() 逐个获取值
16
print(next(gen))  # 5
17
print(next(gen))  # 4
18
print(next(gen))  # 3

1
def simple_gen():
2
    print("开始")
3
    yield 1
4
    print("暂停后恢复")
5
    yield 2
6
    print("再暂停")
7
    yield 3
8
    print("结束")
9

10
gen = simple_gen()
11

12
# 第一次调用 next()：执行到第一个 yield
13
result = next(gen)  # 输出 "开始"
14
print(result)        # 1
15

16
# 第二次调用 next()：从暂停处继续，执行到第二个 yield
17
result = next(gen)  # 输出 "暂停后恢复"
18
print(result)        # 2
19

20
# 第三次调用 next()
21
result = next(gen)  # 输出 "再暂停"
22
print(result)        # 3
23

24
# 第四次调用：函数执行完毕，抛出 StopIteration
25
next(gen)  # 输出 "结束"，然后 StopIteration

1
# 生成器只在需要时才计算值
2
def fibonacci():
3
    """无限斐波那契数列生成器"""
4
    a, b = 0, 1
5
    while True:
6
        yield b
7
        a, b = b, a + b
8

9
# 无限序列也不会耗尽内存
10
fib = fibonacci()
11

12
# 按需获取前 10 个
13
for _ in range(10):
14
    print(next(fib), end=' ')
15
# 输出: 1 1 2 3 5 8 13 21 34 55
16

17
# 内存对比：生成大序列
18
import sys
19

20
# 列表方式（占用大量内存）
21
big_list = list(range(1_000_000))
22
print(sys.getsizeof(big_list))  # ~8MB
23

24
# 生成器方式（几乎不占内存）
25
big_gen = (x for x in range(1_000_000))
26
print(sys.getsizeof(big_gen))   # ~128 bytes

1
# 列表推导式（立即计算，占用内存）
2
squares_list = [x ** 2 for x in range(10)]
3
print(squares_list)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
4

5
# 生成器表达式（惰性求值）
6
squares_gen = (x ** 2 for x in range(10))
7
print(squares_gen)  # <generator object <genexpr> at 0x...>
8

9
# 可以迭代
10
for square in squares_gen:
11
    print(square, end=' ')
12
# 0 1 4 9 16 25 36 49 64 81
13

14
# 生成器表达式的常见用法
15
# 1. 作为函数参数（不需要额外括号）
16
print(sum(x ** 2 for x in range(10)))  # 285
17
print(max(x ** 2 for x in range(10)))  # 81
18

19
# 2. 链式操作
20
nums = (x for x in range(10))
21
evens = (x for x in nums if x % 2 == 0)
22
doubles = (x * 2 for x in evens)
23

24
# 只有在迭代时才真正计算
25
print(list(doubles))  # [0, 4, 8, 12, 16]

1
# yield from 将可迭代对象的所有元素逐个产出
2
def chain(*iterables):
3
    for it in iterables:
4
        yield from it
5

6
# 等价于：
7
def chain_old(*iterables):
8
    for it in iterables:
9
        for item in it:
10
            yield item
11

12
# 使用
13
print(list(chain([1, 2], [3, 4], [5, 6])))
14
# [1, 2, 3, 4, 5, 6]

1
# 扁平化嵌套列表
2
def flatten(nested):
3
    for item in nested:
4
        if isinstance(item, list):
5
            yield from flatten(item)  # 递归调用
6
        else:
7
            yield item
8

9
nested = [1, [2, [3, 4], 5], 6, [7, 8]]
10
print(list(flatten(nested)))
11
# [1, 2, 3, 4, 5, 6, 7, 8]
12

13
# 遍历目录树
14
import os
15

16
def walk_files(root):
17
    for entry in os.scandir(root):
18
        if entry.is_file():
19
            yield entry.path
20
        elif entry.is_dir():
21
            yield from walk_files(entry.path)
22

23
# 遍历指定目录下所有文件
24
# for file in walk_files('/path/to/dir'):
25
#     print(file)

1
# yield from 建立调用者与子生成器之间的双向通道
2
def sub_gen():
3
    """子生成器"""
4
    val = yield "子生成器启动"
5
    print(f"子生成器收到: {val}")
6
    val2 = yield f"第一次收到: {val}"
7
    print(f"子生成器再次收到: {val2}")
8
    return "子生成器结束返回"
9

10
def main_gen():
11
    """主生成器，使用 yield from 委托子生成器"""
12
    result = yield from sub_gen()
13
    print(f"主生成器收到子生成器返回值: {result}")
14
    yield result
15

16
# 使用
17
gen = main_gen()
18
print(next(gen))          # "子生成器启动"
19
print(gen.send("你好"))    # "子生成器收到: 你好"，然后 "第一次收到: 你好"
20
print(gen.send("再见"))    # "子生成器再次收到: 再见"，然后 "主生成器收到子生成器返回值: 子生成器结束返回"

1
def inner():
2
    yield 1
3
    yield 2
4
    return "inner done"
5

6
def outer():
7
    # result 获取 inner 的 return 值
8
    result = yield from inner()
9
    print(f"inner 返回: {result}")
10
    yield 3
11

12
print(list(outer()))
13
# 输出 "inner 返回: inner done"
14
# 输出: [1, 2, 3]

1
def echo():
2
    """回显生成器：接收并返回发送的值"""
3
    msg = "初始值"
4
    while True:
5
        incoming = yield msg  # yield 返回值就是 send() 发送的值
6
        if incoming is None:
7
            break
8
        msg = f"收到: {incoming}"
9

10
gen = echo()
11

12
# 第一次必须用 next() 启动（或 send(None)）
13
print(next(gen))  # "初始值"
14

15
# 使用 send() 发送数据
16
print(gen.send("Hello"))  # "收到: Hello"
17
print(gen.send("World"))  # "收到: World"
18

19
# 发送 None 结束
20
# gen.send(None)  # 抛出 StopIteration

1
def guarded_gen():
2
    try:
3
        yield 1
4
        yield 2
5
        yield 3
6
    except ValueError:
7
        print("捕获到 ValueError")
8
        yield "错误已处理"
9

10
gen = guarded_gen()
11
print(next(gen))  # 1
12
print(gen.throw(ValueError("出错了")))  # 输出 "捕获到 ValueError"，返回 "错误已处理"

1
def resource_gen():
2
    print("打开资源")
3
    try:
4
        yield 1
5
        yield 2
6
        yield 3
7
    finally:
8
        print("关闭资源")
9

10
gen = resource_gen()
11
print(next(gen))  # "打开资源"，1
12
gen.close()       # "关闭资源"
13

14
# 关闭后再调用 next() 会抛出 StopIteration
15
# next(gen)  # StopIteration

1
import itertools
2

3
# count(start, step)：无限计数
4
counter = itertools.count(10, 2)  # 从 10 开始，步长 2
5
print(next(counter))  # 10
6
print(next(counter))  # 12
7
print(next(counter))  # 14
8

9
# cycle(iterable)：无限循环
10
cyc = itertools.cycle(['A', 'B', 'C'])
11
print(next(cyc))  # A
12
print(next(cyc))  # B
13
print(next(cyc))  # C
14
print(next(cyc))  # A
15

16
# repeat(obj, times)：重复
17
repeater = itertools.repeat("Hello", 3)
18
print(list(repeater))  # ['Hello', 'Hello', 'Hello']

1
import itertools
2

3
# product：笛卡尔积
4
for a, b in itertools.product([1, 2], ['x', 'y']):
5
    print(f"{a}-{b}", end=' ')
6
# 1-x 1-y 2-x 2-y
7

8
# permutations：排列（顺序重要）
9
for p in itertools.permutations([1, 2, 3], 2):
10
    print(p, end=' ')
11
# (1, 2) (1, 3) (2, 1) (2, 3) (3, 1) (3, 2)
12

13
# combinations：组合（顺序不重要）
14
for c in itertools.combinations([1, 2, 3], 2):
15
    print(c, end=' ')
16
# (1, 2) (1, 3) (2, 3)
17

18
# combinations_with_replacement：可重复组合
19
for c in itertools.combinations_with_replacement([1, 2], 2):
20
    print(c, end=' ')
21
# (1, 1) (1, 2) (2, 2)

1
import itertools
2

3
# accumulate：累积
4
nums = [1, 2, 3, 4, 5]
5
print(list(itertools.accumulate(nums)))  # [1, 3, 6, 10, 15]
6

7
# chain：连接多个可迭代对象
8
print(list(itertools.chain([1, 2], [3, 4], [5])))  # [1, 2, 3, 4, 5]
9

10
# chain.from_iterable：从嵌套可迭代对象扁平化
11
print(list(itertools.chain.from_iterable([[1, 2], [3, 4]])))  # [1, 2, 3, 4]
12

13
# compress：根据选择器过滤
14
data = ['A', 'B', 'C', 'D']
15
selectors = [True, False, True, False]
16
print(list(itertools.compress(data, selectors)))  # ['A', 'C']
17

18
# dropwhile：丢弃满足条件的前缀元素
19
nums = [1, 3, 5, 2, 4, 6]
20
print(list(itertools.dropwhile(lambda x: x < 5, nums)))  # [5, 2, 4, 6]
21

22
# takewhile：取满足条件的前缀元素
23
print(list(itertools.takewhile(lambda x: x < 5, nums)))  # [1, 3]
24

25
# islice：切片（支持步长）
26
print(list(itertools.islice(range(10), 2, 8, 2)))  # [2, 4, 6]
27

28
# starmap：将参数元组解包后传入函数
29
points = [(1, 2), (3, 4), (5, 6)]
30
print(list(itertools.starmap(lambda x, y: x + y, points)))  # [3, 7, 11]

1
import itertools
2

3
# groupby：按键分组（注意：需要先排序）
4
data = [('A', 1), ('A', 2), ('B', 3), ('B', 4), ('A', 5)]
5
# 按第一个元素分组
6
for key, group in itertools.groupby(data, key=lambda x: x[0]):
7
    print(f"{key}: {list(group)}")
8
# A: [('A', 1), ('A', 2)]
9
# B: [('B', 3), ('B', 4)]
10
# A: [('A', 5)]  （注意：相同键不连续时会分成多个组）
11

12
# tee：从一个迭代器创建多个独立迭代器
13
nums = range(5)
14
it1, it2 = itertools.tee(nums, 2)
15
print(list(it1))  # [0, 1, 2, 3, 4]
16
print(list(it2))  # [0, 1, 2, 3, 4]
17

18
# zip_longest：不等长 zip，用 fillvalue 填充
19
a = [1, 2, 3]
20
b = ['x', 'y']
21
print(list(itertools.zip_longest(a, b, fillvalue='-')))
22
# [(1, 'x'), (2, 'y'), (3, '-')]

1
def read_large_file(filepath):
2
    """逐行读取大文件，内存占用极低"""
3
    with open(filepath, 'r', encoding='utf-8') as f:
4
        for line in f:
5
            yield line.strip()
6

7
# 使用
8
# for line in read_large_file('huge_file.txt'):
9
#     process(line)

1
def gen_numbers():
2
    """生成数字"""
3
    for i in range(1000000):
4
        yield i
5

6
def filter_even(numbers):
7
    """过滤偶数"""
8
    for n in numbers:
9
        if n % 2 == 0:
10
            yield n
11

12
def square(numbers):
13
    """平方"""
14
    for n in numbers:
15
        yield n * n
16

17
def take_first(numbers, n):
18
    """取前 n 个"""
19
    for i, num in enumerate(numbers):
20
        if i >= n:
21
            break
22
        yield num
23

24
# 构建管道
25
pipeline = take_first(square(filter_even(gen_numbers())), 5)
26

27
# 惰性求值：只计算需要的
28
print(list(pipeline))  # [0, 4, 16, 36, 64]

1
def fetch_all_pages(api_func, start_page=1, page_size=100):
2
    """分页获取所有数据"""
3
    page = start_page
4
    while True:
5
        results = api_func(page, page_size)
6
        if not results:
7
            break
8
        for item in results:
9
            yield item
10
        page += 1
11
        if len(results) < page_size:
12
            break
13

14
# 模拟 API
15
def mock_api(page, page_size):
16
    if page > 3:
17
        return []
18
    return [f"item_{page}_{i}" for i in range(page_size)]
19

20
# 使用
21
all_items = fetch_all_pages(mock_api)
22
for item in all_items:
23
    pass  # 处理每个 item

1
def traffic_light():
2
    """交通灯状态机"""
3
    while True:
4
        # 红灯
5
        yield "🔴 红灯 - 停止"
6
        # 绿灯
7
        yield "🟢 绿灯 - 通行"
8
        # 黄灯
9
        yield "🟡 黄灯 - 注意"
10

11
light = traffic_light()
12
for _ in range(6):
13
    print(next(light))
14
# 🔴 红灯 - 停止
15
# 🟢 绿灯 - 通行
16
# 🟡 黄灯 - 注意
17
# 🔴 红灯 - 停止
18
# ...

1
def get_numbers():
2
    yield from [1, 2, 3]
3

4
gen = get_numbers()
5
print(list(gen))  # [1, 2, 3]
6
print(list(gen))  # []  （生成器已耗尽）
7

8
# ✅ 解决方法：每次需要时重新创建生成器
9
print(list(get_numbers()))  # [1, 2, 3]
10
print(list(get_numbers()))  # [1, 2, 3]

1
# ❌ 错误：在函数中使用生成器表达式需要注意
2
print(sum((x ** 2 for x in range(10))))  # 可以，但多一层括号
3

4
# ✅ 正确：作为唯一参数时可以省略括号
5
print(sum(x ** 2 for x in range(10)))  # 285

1
# ❌ 错误：迭代时修改列表会导致意外行为
2
def remove_even(lst):
3
    for item in lst:
4
        if item % 2 == 0:
5
            lst.remove(item)  # 迭代时修改列表
6
    return lst
7

8
nums = [1, 2, 3, 4, 5, 6]
9
print(remove_even(nums))  # [1, 3, 5]  （可能跳过元素）
10

11
# ✅ 正确：创建新列表或使用列表推导式
12
nums = [1, 2, 3, 4, 5, 6]
13
result = [x for x in nums if x % 2 != 0]

1
# ❌ 错误：直接 send 非 None 值
2
gen = echo()
3
# gen.send("Hello")  # TypeError: can't send non-None value to a just-started generator
4

5
# ✅ 正确：先用 next() 或 send(None) 启动
6
gen = echo()
7
next(gen)        # 启动
8
gen.send("Hi")   # 正常

1
✅ 处理大文件/大数据集 → 惰性求值节省内存
2
✅ 无限序列 → 斐波那契、计数器
3
✅ 数据流管道 → 多步转换、链式操作
4
✅ 分页 API → 逐页加载
5
✅ 状态机 → 保存状态、分步执行
6
✅ 协程基础 → send/throw/close 双向通信
7

8
❌ 需要随机访问 → 使用列表
9
❌ 需要多次迭代 → 使用可迭代对象（每次返回新迭代器）
10
❌ 数据量很小 → 直接用列表更简单