迭代器与生成器

迭代（Iteration）是编程中最基础也最重要的操作之一。Python 通过迭代器协议和生成器提供了强大而优雅的迭代机制。理解迭代器和生成器，不仅能让你写出更高效的代码，还能掌握处理大数据集和无限序列的能力。

迭代基础

可迭代对象 vs 迭代器

理解这两个概念是掌握 Python 迭代的基础：

from typing import Iterable, Iterator


# 可迭代对象（Iterable）：可以用于 for 循环的对象
# 实现了 __iter__() 方法，返回一个迭代器

# 迭代器（Iterator）：负责产生数据的对象
# 实现了 __iter__() 和 __next__() 方法

# 常见可迭代对象
print(isinstance([], Iterable))   # True
print(isinstance({}, Iterable))   # True
print(isinstance("hello", Iterable))  # True
print(isinstance(range(10), Iterable))  # True

# 列表是可迭代对象，但不是迭代器
print(isinstance([], Iterator))    # False

for 循环内部原理

# Python 的 for 循环等价于以下过程：

# 你写的代码
for item in [1, 2, 3]:
    print(item)

# Python 实际执行的代码
_iter = iter([1, 2, 3])  # 获取迭代器
while True:
    try:
        item = next(_iter)  # 获取下一个值
        print(item)
    except StopIteration:   # 没有更多元素
        break

理解这个机制很重要——它解释了为什么迭代器是惰性的（按需生成数据）以及为什么可迭代对象可以用于 for 循环。

Iterable 和 Iterator 的严格定义

from typing import Iterable, Iterator
from collections.abc import Iterable as ABCIterable, Iterator as ABCIterator


# 可迭代对象：实现了 __iter__()
# 迭代器：实现了 __iter__() 和 __next__()

class MyIterable:
    """自定义可迭代对象"""
    def __iter__(self):
        return MyIterator([1, 2, 3])

class MyIterator:
    """自定义迭代器"""
    def __init__(self, data):
        self._data = data
        self._index = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self._index >= len(self._data):
            raise StopIteration
        value = self._data[self._index]
        self._index += 1
        return value


obj = MyIterable()
print(isinstance(obj, ABCIterable))  # True
print(hasattr(obj, "__iter__"))      # True

it = iter(obj)
print(isinstance(it, ABCIterator))   # True

创建自定义迭代器

类方式实现

class Squares:
    """生成前 n 个平方数的迭代器"""

    def __init__(self, limit: int):
        self.limit = limit
        self.current = 0

    def __iter__(self):
        """返回迭代器本身"""
        self.current = 0
        return self

    def __next__(self) -> int:
        self.current += 1
        if self.current > self.limit:
            raise StopIteration
        return self.current ** 2


# 使用
squares = Squares(5)
for n in squares:
    print(n, end=" ")  # 1 4 9 16 25

# 可以重新迭代
print()
for n in squares:
    print(n, end=" ")  # 1 4 9 16 25

可迭代对象与迭代器分离

更好的设计是让可迭代对象每次返回一个新的迭代器：

class FibonacciSequence:
    """斐波那契数列 -- 每次迭代返回新迭代器"""

    def __init__(self, max_count: int = 10):
        self.max_count = max_count

    def __iter__(self):
        return FibonacciIterator(self.max_count)


class FibonacciIterator:
    """斐波那契数列的迭代器"""

    def __init__(self, max_count: int):
        self.max_count = max_count
        self.count = 0
        self.a, self.b = 0, 1

    def __iter__(self):
        return self

    def __next__(self) -> int:
        if self.count >= self.max_count:
            raise StopIteration
        self.count += 1
        self.a, self.b = self.b, self.a + self.b
        return self.a


fib = FibonacciSequence(8)
for n in fib:
    print(n, end=" ")  # 1 1 2 3 5 8 13 21

print()
for n in fib:  # 返回新的迭代器，重新开始
    print(n, end=" ")  # 1 1 2 3 5 8 13 21

生成器函数

生成器是 Python 中最优雅的迭代器创建方式。使用 yield 关键字，一个普通函数就变成了生成器：

from typing import Iterator


def simple_generator():
    """最简单的生成器"""
    yield 1
    yield 2
    yield 3


gen = simple_generator()
print(type(gen))  # <class 'generator'>
print(isinstance(gen, Iterator))  # True

print(next(gen))  # 1
print(next(gen))  # 2
print(next(gen))  # 3

# 可以使用 for 循环
for value in simple_generator():
    print(value, end=" ")  # 1 2 3

yield 的工作机制

yield 与 return 的关键区别：

return：返回值并退出函数
yield：暂停函数，保存所有状态，返回一个值；下次调用 next() 时从暂停处继续

def stateful_gen():
    print("生成器启动")
    value = 1
    print(f"第一次 yield，value = {value}")
    yield value

    value = 2
    print(f"第二次 yield，value = {value}")
    yield value

    value = 3
    print(f"第三次 yield，value = {value}")
    yield value

    print("生成器结束")


gen = stateful_gen()
print("调用 next(gen) 第一次")
result = next(gen)
print(f"得到: {result}")

print("调用 next(gen) 第二次")
result = next(gen)
print(f"得到: {result}")

print("调用 next(gen) 第三次")
result = next(gen)
print(f"得到: {result}")

输出解释了这个机制——生成器函数在执行到 yield 时暂停，下次调用时从暂停点继续。

用生成器重写迭代器

# 生成器版本的 Fibonacci
def fibonacci(max_count: int):
    """斐波那契数列生成器"""
    a, b = 0, 1
    count = 0
    while count < max_count:
        a, b = b, a + b
        count += 1
        yield a


for n in fibonacci(8):
    print(n, end=" ")  # 1 1 2 3 5 8 13 21


# 生成器版本的 Squares
def squares(limit: int):
    for i in range(1, limit + 1):
        yield i ** 2


print(list(squares(5)))  # [1, 4, 9, 16, 25]

生成器表达式

生成器表达式（Generator Expression）是列表推导式的惰性版本——它不立即创建列表，而是返回一个生成器对象：

# 列表推导式 -- 立即计算所有值，占用内存
list_squares = [x ** 2 for x in range(10)]
print(list_squares)       # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
print(type(list_squares)) # <class 'list'>

# 生成器表达式 -- 惰性求值，逐个生成
gen_squares = (x ** 2 for x in range(10))
print(gen_squares)        # <generator object <genexpr> at 0x...>
print(type(gen_squares))  # <class 'generator'>

for n in gen_squares:
    print(n, end=" ")     # 0 1 4 9 16 25 36 49 64 81

生成器表达式 vs 列表推导式

import sys

# 内存对比
list_data = [x for x in range(1_000_000)]
gen_data = (x for x in range(1_000_000))

print(f"列表大小: {sys.getsizeof(list_data) / 1024 / 1024:.2f} MB")
# 列表大小: ~7.63 MB

print(f"生成器大小: {sys.getsizeof(gen_data)} bytes")
# 生成器大小: ~208 bytes


# 适用场景对比

# 场景 1：只需要迭代一次 -- 生成器
total = sum(x ** 2 for x in range(10_000_000))  # 高效

# 场景 2：需要多次访问 -- 列表
data = [x ** 2 for x in range(1000)]
for _ in range(3):
    for val in data:  # 可以多次迭代
        pass

# 场景 3：需要随机访问 -- 列表
data = [x ** 2 for x in range(1000)]
print(data[42])  # 随机访问

何时选择生成器表达式

# 适合生成器表达式
sum(x ** 2 for x in range(1000))
max(len(line) for line in file)
any(x > 0 for x in data)
all(isinstance(x, int) for x in data)

# 适合列表推导式
[x * 2 for x in data]       # 需要多次使用
[x for x in data if x > 0]  # 需要切片或索引

生成器的高级方法

生成器除了 __next__() 外，还有三个重要方法：

send(value)

向生成器内部发送值，并返回下一个 yield 的值：

def echo_processor():
    """回声处理器 -- 接收并处理值"""
    print("生成器启动")
    while True:
        received = yield  # 不产生值，只接收
        print(f"收到: {received}")


gen = echo_processor()
next(gen)               # 启动生成器，执行到第一个 yield

gen.send("Hello")       # 输出：收到: Hello
gen.send("World")       # 输出：收到: World
gen.close()             # 关闭生成器

send 的实际应用

def running_average():
    """运行平均值计算器"""
    total = 0.0
    count = 0
    average = None

    while True:
        value = yield average
        if value is None:
            continue
        total += value
        count += 1
        average = total / count


avg = running_average()
next(avg)  # 启动到第一个 yield

print(avg.send(10))   # 10.0
print(avg.send(20))   # 15.0
print(avg.send(30))   # 20.0
print(avg.send(100))  # 40.0

throw(exception)

向生成器中注入异常：

def safe_divide():
    """安全除法生成器"""
    try:
        x = yield
        y = yield
        result = x / y
        yield result
    except ZeroDivisionError:
        yield float("inf")
    except Exception as e:
        yield f"Error: {e}"

close()

关闭生成器，使其后续的 __next__() 抛出 StopIteration：

def infinite_gen():
    i = 0
    while True:
        yield i
        i += 1


gen = infinite_gen()
print(next(gen))  # 0
print(next(gen))  # 1
gen.close()

yield from 委托

yield from 允许一个生成器将部分工作委托给另一个生成器（或可迭代对象）：

def generator_a():
    yield "A1"
    yield "A2"

def generator_b():
    yield "B1"
    yield "B2"

def combined():
    yield from generator_a()  # 委托给 A
    yield from generator_b()  # 委托给 B
    yield "Final"


print(list(combined()))
# ['A1', 'A2', 'B1', 'B2', 'Final']


# 等价于：
def combined_manual():
    for item in generator_a():
        yield item
    for item in generator_b():
        yield item
    yield "Final"

yield from 的经典应用

def flatten(nested_list):
    """展平嵌套列表（不限层级）"""
    for item in nested_list:
        if isinstance(item, list):
            yield from flatten(item)  # 递归展开
        else:
            yield item


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

无限序列与惰性求值

生成器非常适合处理无限序列：

def natural_numbers():
    """自然数序列（无限）"""
    n = 1
    while True:
        yield n
        n += 1


def even_numbers():
    """偶数序列（无限）"""
    for n in natural_numbers():
        if n % 2 == 0:
            yield n


# 只取需要的值
first_10_even = []
for i, n in enumerate(even_numbers()):
    if i >= 10:
        break
    first_10_even.append(n)

print(first_10_even)  # [2, 4, 6, 8, 10, 12, 14, 16, 18, 20]


# 使用 itertools.islice 简化
from itertools import islice

print(list(islice(even_numbers(), 10)))
# [2, 4, 6, 8, 10, 12, 14, 16, 18, 20]

惰性求值的优势

def read_large_file(filename: str):
    """惰性读取大文件 -- 一次只读一行"""
    with open(filename) as f:
        for line in f:
            yield line.strip()


# 只处理前 10 行
for i, line in enumerate(read_large_file("huge_log.txt")):
    if i >= 10:
        break
    process(line)


# 链式惰性处理
from itertools import islice

def numbers():
    n = 1
    while True:
        yield n
        n += 1

# 取前 20 个平方数中大于 50 的数
result = list(islice(
    (x ** 2 for x in numbers() if x ** 2 > 50),
    20
))
print(result[:5])  # [64, 81, 100, 121, 144]

综合实战

实战 1：读取大文件并处理

import csv
from typing import Iterator


def read_csv_chunks(filepath: str, chunk_size: int = 1000) -> Iterator[list[dict]]:
    """按块读取大 CSV 文件"""
    with open(filepath, newline="") as f:
        reader = csv.DictReader(f)
        chunk = []
        for row in reader:
            chunk.append(row)
            if len(chunk) >= chunk_size:
                yield chunk
                chunk = []
        if chunk:
            yield chunk


def extract_emails(filepath: str) -> Iterator[str]:
    """惰性提取所有邮箱地址"""
    import re
    with open(filepath) as f:
        for line in f:
            emails = re.findall(r"[\w.%+-]+@[\w.-]+\.[a-zA-Z]{2,}", line)
            yield from emails

实战 2：实现管道处理

def read_lines(filepath: str) -> Iterator[str]:
    with open(filepath) as f:
        yield from f


def strip_whitespace(lines: Iterator[str]) -> Iterator[str]:
    for line in lines:
        yield line.strip()


def skip_empty(lines: Iterator[str]) -> Iterator[str]:
    for line in lines:
        if line:
            yield line


def skip_comments(lines: Iterator[str]) -> Iterator[str]:
    for line in lines:
        if not line.startswith("#"):
            yield line


def extract_urls(lines: Iterator[str]) -> Iterator[str]:
    import re
    url_pattern = r"https?://[^\s]+"
    for line in lines:
        for url in re.findall(url_pattern, line):
            yield url


# 管道组装
pipeline = extract_urls(
    skip_comments(
        skip_empty(
            strip_whitespace(
                read_lines("data.txt")
            )
        )
    )
)

# 获取前 10 个 URL
from itertools import islice
for url in islice(pipeline, 10):
    print(url)

实战 3：流式数据处理

from collections import deque
from itertools import islice
from typing import Iterator


def sliding_window(stream: Iterator, window_size: int) -> Iterator[tuple]:
    """滑动窗口生成器"""
    window = deque(islice(stream, window_size), maxlen=window_size)
    if len(window) < window_size:
        return
    yield tuple(window)
    for item in stream:
        window.append(item)
        yield tuple(window)


def moving_average(data: Iterator[float], window: int = 3) -> Iterator[float]:
    """移动平均"""
    for window_values in sliding_window(data, window):
        yield sum(window_values) / window


# 使用
data = iter([10, 12, 15, 14, 18, 20, 22, 25])
for avg in moving_average(data, 3):
    print(f"{avg:.2f}", end=" ")  # 12.33 13.67 15.67 17.33 20.00 22.33

实战 4：分页 API 请求

from typing import Iterator
import time


def paginated_api(api_url: str, page_size: int = 100) -> Iterator[dict]:
    """自动翻页的 API 数据获取"""
    page = 1
    while True:
        print(f"请求第 {page} 页...")
        data = mock_api_request(api_url, page=page, page_size=page_size)

        if not data["results"]:
            break

        for item in data["results"]:
            yield item

        if not data.get("has_next"):
            break

        page += 1
        time.sleep(0.5)


def mock_api_request(url: str, page: int, page_size: int) -> dict:
    """模拟 API 响应"""
    start = (page - 1) * page_size
    results = [{"id": i, "name": f"Item {i}"}
               for i in range(start, start + min(page_size, 25))]
    return {
        "results": results,
        "has_next": page < 3,
    }


# 惰性处理所有数据
total = 0
for item in paginated_api("https://api.example.com/items"):
    print(f"  处理: {item['name']}")
    total += 1

print(f"总共处理: {total} 条数据")

实战 5：令牌桶限流器

def token_bucket(rate: float, capacity: int):
    """令牌桶限流器"""
    import time
    tokens = capacity
    last_refill = time.perf_counter()

    while True:
        now = time.perf_counter()
        elapsed = now - last_refill
        tokens = min(capacity, tokens + elapsed * rate)
        last_refill = now

        requested = yield tokens >= 1
        if requested:
            tokens -= 1
            yield True
        else:
            yield False


import time

limiter = token_bucket(rate=3, capacity=3)  # 每秒 3 个请求，桶容量 3
next(limiter)  # 启动

for i in range(10):
    allowed = limiter.send(True)
    if allowed:
        print(f"请求 {i+1}: 允许")
    else:
        print(f"请求 {i+1}: 被限流")
    time.sleep(0.1)

常见陷阱

陷阱 1：生成器只能迭代一次

gen = (x ** 2 for x in range(5))

print(list(gen))  # [0, 1, 4, 9, 16]
print(list(gen))  # [] -- 已经消费完！

陷阱 2：在迭代时修改数据结构

# 不安全的做法
items = [1, 2, 3, 4, 5]
for item in items:
    if item % 2 == 0:
        items.remove(item)

# 安全的做法：创建新列表
items = [1, 2, 3, 4, 5]
items = [item for item in items if item % 2 != 0]
print(items)  # [1, 3, 5]

陷阱 3：生成器表达式中的闭包问题

# 问题：所有函数都捕获了循环结束后的最终值
funcs = []
for i in range(3):
    funcs.append(lambda: i)

for f in funcs:
    print(f(), end=" ")  # 2 2 2 -- 不是 0 1 2

陷阱 4：StopIteration 在生成器内部被误用

# 错误：在生成器内部手动抛出 StopIteration
def bad_gen():
    raise StopIteration  # Python 3.7+ 中已被弃用
    yield 1

# 正确：使用 return
def good_gen():
    return  # 等于 raise StopIteration
    yield 1

迭代器与内存效率总结

import sys


# 列表：所有数据在内存中
list_squares = [x ** 2 for x in range(100_000)]
print(f"列表内存: {sys.getsizeof(list_squares) / 1024:.1f} KB")

# 生成器：按需产生
gen_squares = (x ** 2 for x in range(100_000))
print(f"生成器内存: {sys.getsizeof(gen_squares)} bytes")

# 性能对比
import time

# 列表方式
start = time.perf_counter()
total_list = sum(x ** 2 for x in range(10_000_000))
print(f"生成器方式: {time.perf_counter() - start:.3f}s")

# 生成器方式
start = time.perf_counter()
total_gen = sum(x ** 2 for x in range(10_000_000))
print(f"生成器方式: {time.perf_counter() - start:.3f}s")

小结

本篇全面学习了 Python 的迭代器与生成器机制：

可迭代对象 vs 迭代器：Iterable（可被 for 循环）和 Iterator（迭代器对象）的区别
for 循环的底层原理：iter() + next() + StopIteration 的三步曲
自定义迭代器：通过 __iter__() 和 __next__() 创建自定义迭代器类
生成器函数：使用 yield 创建更简洁的迭代器
生成器表达式：列表推导式的惰性版本，节省内存
生成器高级方法：send()、throw()、close() 控制生成器行为
yield from：委托给子生成器，简化代码
无限序列：生成器天然适合表示无穷数据流
惰性求值：只在需要时才计算，节省内存和时间

生成器是 Python 中最优雅的特性之一。合理使用可以显著改善代码的内存效率和可读性。

Summary: 迭代器协议、生成器、yield、yield from、惰性求值