Python生成器(Generator)与协程深度解析：从迭代器协议到yield表达式、子生成器委托与异步生成器实战

Python生成器是语言中最优雅且强大的特性之一。它不仅提供了一种惰性求值的数据生产方式，更是Python协程机制的基石。从简单的迭代器替代到复杂的异步IO框架，生成器的身影无处不在。然而，许多开发者对生成器的理解停留在

关键字的表面用法，对其底层机制、

与

的双向通信、

的子生成器委托，以及从生成器到原生协程的演进历程缺乏系统认识。本文将从迭代器协议出发，逐层深入，带你全面掌握Python生成器的核心原理与实战技巧。

一、迭代器协议：生成器的根基

理解生成器的第一步，是理解Python的迭代器协议。在Python中，任何实现了

和

方法的对象都是迭代器。

返回迭代器对象自身，

返回下一个值，当没有更多元素时抛出

异常。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class Countdown:

    def __init__(self, start):

        self.current = start



    def __iter__(self):

        return self



    def __next__(self):

        if self.current <= 0:

            raise StopIteration

        self.current -= 1

        return self.current + 1



# 使用方式

for num in Countdown(5):

    print(num)  # 5, 4, 3, 2, 1

手动实现迭代器需要编写大量样板代码，尤其是

和

的声明、状态管理和

的抛出。生成器正是为了简化这一过程而诞生的——它让Python解释器自动为你生成迭代器。

二、生成器函数与yield表达式

包含

关键字的函数不再是普通函数，而是一个生成器工厂。调用它不会执行函数体，而是返回一个生成器对象。生成器对象自动实现了迭代器协议，每次调用

时，函数体从上一次

的位置恢复执行，直到遇到下一个

或函数结束。

1
2
3
4
5
6
7
8
9
10
11
def countdown(start):

    current = start

    while current > 0:

        yield current

        current -= 1



gen = countdown(5)

print(type(gen))          # <class 'generator'>

print(next(gen))          # 5

print(next(gen))          # 4

print(list(gen))          # [3, 2, 1]

2.1 生成器的内部状态保存

生成器的核心魔法在于挂起与恢复。当执行到

时，生成器会保存当前的执行帧（包括局部变量、指令指针和异常状态），将控制权交还给调用者。下次

时，从保存的状态精确恢复。这一机制使得生成器天然适合处理流式数据和无限序列。

1
2
3
4
5
6
7
8
9
10
def fibonacci():

    a, b = 0, 1

    while True:

        yield a

        a, b = b, a + b



# 惰性求值：只计算需要的值

fib = fibonacci()

first_ten = [next(fib) for _ in range(10)]

print(first_ten)  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

2.2 生成器表达式

与列表推导式类似，生成器表达式提供了一种简洁的生成器创建方式，使用圆括号而非方括号。它的惰性求值特性使得内存占用从O(n)降为O(1)。

1
2
3
4
5
6
7
8
9
10
11
12
13
import sys



# 列表推导式：一次性生成所有元素

nums_list = [x ** 2 for x in range(1_000_000)]

print(sys.getsizeof(nums_list))  # ~8.5MB



# 生成器表达式：惰性生成

nums_gen = (x ** 2 for x in range(1_000_000))

print(sys.getsizeof(nums_gen))   # ~200 bytes



# 生成器表达式可直接传入sum/max/min等函数

total = sum(x ** 2 for x in range(100))

print(total)  # 328350

三、双向通信：send()、throw()与close()

生成器不仅是数据的生产者，还可以是数据的消费者。

实际上是一个表达式，它可以接收外部通过

传入的值。这使得生成器具备了协程的雏形。

3.1 send()方法与值传递

将

发送到生成器内部，作为当前

表达式的返回值，然后生成器恢复执行到下一个

。注意：第一次调用

（或者先用

启动），因为生成器还没有执行到

表达式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def accumulator():

    total = 0

    while True:

        value = yield total  # yield产出total，同时接收send传入的value

        if value is None:

            break

        total += value



gen = accumulator()

next(gen)            # 启动生成器，执行到第一个yield，产出0

print(gen.send(10))  # 发送10，yield表达式返回10，total=10，产出10

print(gen.send(20))  # 发送20，yield表达式返回20，total=30，产出30

print(gen.send(5))   # 发送5，total=35，产出35

gen.close()          # 关闭生成器

3.2 throw()方法与异常注入

在生成器挂起的

位置抛出指定异常。这为生成器提供了从外部中断或信号通知的能力。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def resilient_processor():

    while True:

        try:

            data = yield

            print(f"Processing: {data}")

        except ValueError as e:

            print(f"Skipping bad data: {e}")

        except GeneratorExit:

            print("Cleaning up resources...")

            raise



gen = resilient_processor()

next(gen)                             # 启动

gen.send("normal data")                # Processing: normal data

gen.throw(ValueError, "bad format")    # Skipping bad data: bad format

gen.send("resumed")                    # Processing: resumed

gen.close()                           # Cleaning up resources...

3.3 close()方法与清理

在生成器内部抛出

异常。如果生成器捕获了该异常，应执行清理逻辑后重新抛出或直接返回。忽略

会导致

。

四、yield from：子生成器委托

Python 3.3引入的

是生成器演进的关键一步。它不仅简化了生成器嵌套的写法，更建立了主生成器与子生成器之间的透明双向通道——

、

和

调用会自动穿透到子生成器。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def sub_gen():

    received = yield "Sub started"

    yield f"Sub received: {received}"

    return "Sub result"



def main_gen():

    # yield from建立双向通道

    result = yield from sub_gen()

    yield f"Main got sub result: {result}"



gen = main_gen()

print(next(gen))          # Sub started

print(gen.send("from main"))  # Sub received: from main

print(next(gen))          # Main got sub result: Sub result

4.1 yield from的核心语义

的行为可以概括为：

• 迭代阶段：自动从子生成器获取值并向上产出，相当于一个

  1	for item in sub: yield item

  的简化

• 值传递：调用者

  1

  send()

  的值直接传递给子生成器的

  1

  yield

  表达式

• 异常传递：调用者

  1

  throw()

  的异常直接在子生成器的

  1

  yield

  处抛出

• 返回值捕获：子生成器的

  1

  return

  值成为

  1	yield from

  表达式的值

这意味着

不仅仅是语法糖，它建立了一个语义等价的委托关系——对于调用者而言，主生成器和子生成器就像一个统一的生成器。

4.2 实战：用yield from构建数据处理管道

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def read_lines(filepath):

    with open(filepath, 'r', encoding='utf-8') as f:

        for line in f:

            stripped = yield line.strip()



def filter_comments(source):

    for line in source:

        if not line.startswith('#') and line:

            yield line



def parse_config(source):

    lines = yield from read_lines(source)

    for line in filter_comments(lines):

        if '=' in line:

            key, value = line.split('=', 1)

            yield (key.strip(), value.strip())



# 使用管道

config = dict(parse_config('/etc/app.conf'))

五、从生成器协程到async/await

Python的异步编程经历了三个重要阶段，而生成器是其中的关键纽带。

5.1 早期协程：基于生成器的协程

在Python 3.5引入

之前，开发者利用

和

实现了协程。通过

装饰器或

装饰器，生成器函数可以被标记为协程。

1
2
3
4
5
6
7
8
9
10
11
12
13
import asyncio

import types



@types.coroutine

def old_style_coroutine():

    result = yield from asyncio.sleep(1)

    print("Waited 1 second")

    return result



# 现代写法对比

async def modern_coroutine():

    await asyncio.sleep(1)

    print("Waited 1 second")

5.2 原生协程与生成器的区别

Python 3.5引入的

定义的是原生协程，它与生成器有以下关键区别：

关键洞察：

的语义与

高度相似——都是将控制权交给被委托的对象，并等待其完成。

实际上是

在异步上下文中的专有形式，限制只能用于awaitable对象（实现了

方法的对象）。

六、异步生成器：async yield

Python 3.6引入了异步生成器，将

与

结合，使得在异步上下文中也能进行惰性求值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import asyncio

import aiohttp



async def fetch_paginated(api_url, page_size=100):

    page = 1

    async with aiohttp.ClientSession() as session:

        while True:

            async with session.get(

                f"{api_url}?page={page}&size={page_size}"

            ) as resp:

                data = await resp.json()

                if not data['items']:

                    break

                for item in data['items']:

                    yield item

                page += 1



async def process_all():

    async for item in fetch_paginated("https://api.example.com/data"):

        print(f"Processing: {item['id']}")



asyncio.run(process_all())

异步生成器的关键规则：

• 使用

  1

  async for

  迭代，不能用普通

  1

  for

• 使用

  1	async with

  进行异步上下文管理

• 每个

  1

  yield

  前可以有

  1

  await

  调用

• 异步生成器没有

  1	return value

  （返回值会被忽略）

• 使用

  1

  aclose()

  而非

  1

  close()

  关闭

七、生成器实战模式

7.1 流式处理大文件

处理GB级日志文件时，一次性加载到内存是不可行的。生成器提供了一种优雅的流式处理方案：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def parse_log_stream(filepath, pattern=None):

    import re

    regex = re.compile(pattern) if pattern else None

    with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:

        for line_num, line in enumerate(f, 1):

            line = line.strip()

            if not line:

                continue

            if regex and not regex.search(line):

                continue

            parts = line.split(' ', 3)

            if len(parts) >= 4:

                yield {

                    'timestamp': parts[0] + ' ' + parts[1],

                    'level': parts[2].rstrip(':'),

                    'message': parts[3],

                    'line': line_num

                }



# 链式处理

errors = (e for e in parse_log_stream('/var/log/app.log', r'ERROR'))

critical = (e for e in errors if 'CRITICAL' in e['message'])

for entry in critical:

    print(f"[{entry['timestamp']}] {entry['message']}")

7.2 状态机实现

生成器的

方法天然适合实现状态机——每个

就是一个状态节点，

传入的值是状态转移的输入。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def http_parser():

    state = 'METHOD'

    method = path = version = ''

    headers = {}

    body_chunks = []



    while True:

        data = yield

        if data is None:

            continue



        if state == 'METHOD':

            parts = data.split(' ')

            if len(parts) >= 3:

                method, path, version = parts[0], parts[1], parts[2]

                state = 'HEADERS'



        elif state == 'HEADERS':

            if data == '':

                state = 'BODY'

            elif ':' in data:

                key, val = data.split(':', 1)

                headers[key.strip()] = val.strip()



        elif state == 'BODY':

            body_chunks.append(data)



parser = http_parser()

next(parser)

parser.send('GET /api/users HTTP/1.1')

parser.send('Host: example.com')

parser.send('Content-Type: application/json')

parser.send('')

parser.send('{"query": "active"}')

7.3 上下文管理器与生成器

装饰器将生成器函数转换为上下文管理器，这是标准库中最优雅的生成器应用之一：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from contextlib import contextmanager

import time



@contextmanager

def timer(label="Operation"):

    start = time.perf_counter()

    try:

        yield

    finally:

        elapsed = time.perf_counter() - start

        print(f"{label} took {elapsed:.4f}s")



with timer("Data loading"):

    data = load_large_dataset()



with timer("Model training"):

    model.fit(data)

八、生成器性能陷阱与最佳实践

8.1 常见陷阱

陷阱一：生成器只能消费一次。生成器是单次迭代器，消费完毕后再次迭代不会产出任何值。如果需要多次迭代，应使用

物化或

。

1
2
3
4
5
6
7
8
9
10
11
12
13
def numbers():

    yield from range(5)



gen = numbers()

print(list(gen))  # [0, 1, 2, 3, 4]

print(list(gen))  # [] -- already exhausted!



# Solution 1: materialize

data = list(numbers())



# Solution 2: factory function

def numbers_factory():

    return (x for x in range(5))

陷阱二：生成器中的异常被吞噬。当生成器未完全消费时（如在

循环中

），生成器被垃圾回收时会在

处抛出

。如果此时有未处理的异常，可能掩盖真正的错误。

陷阱三：过度嵌套yield from。深度嵌套的

链会增加调用栈深度，在极端情况下影响性能。对于简单的迭代转发，优先考虑

。

8.2 最佳实践

• 明确区分生成器函数与普通函数：一旦函数包含

  1

  yield

  ，它就是生成器工厂。不要在生成器函数中混用

  1	return value

  （Python 3.3+中

  1	return value

  会设置

  1	StopIteration.value

  ，但容易被忽略）。

• 使用类型注解标注生成器：

  1	Generator[YieldType, SendType, ReturnType]

  比

  1	Iterator[YieldType]

  更精确地描述了生成器的完整类型。

• 及时关闭不再使用的生成器：调用

  1

  close()

  确保资源释放，尤其是在生成器持有文件句柄或数据库连接时。

• 优先使用生成器表达式替代列表推导：当结果只需要被消费一次且不需要索引访问时，生成器表达式更节省内存。

1
2
3
4
5
6
7
from typing import Generator



def structured_generator(n: int) -> Generator[int, str, None]:

    for i in range(n):

        message = yield i

        if message:

            print(f"Received: {message}")

九、总结

Python生成器从迭代器协议出发，经历了

表达式、

双向通信、

子生成器委托，最终演进出原生协程

和异步生成器

。这一演进历程体现了Python语言设计的核心哲学——用统一的抽象解决不同层次的问题。

掌握生成器的关键要点：

• 生成器函数是迭代器工厂，

  1

  yield

  实现挂起与恢复

• 1

  send()

  /

  1

  throw()

  /

  1

  close()

  提供双向通信与生命周期控制

• 1	yield from

  建立透明的子生成器委托通道

• 原生协程是生成器协程的演进，

  1

  await

  是

  1	yield from

  的异步版本

• 异步生成器将惰性求值引入异步编程
• 实战中注意生成器的单次消费、异常处理和资源释放

生成器是Python中少有的“小而美”的特性——概念简单，却能组合出极其强大的抽象。无论是流式数据处理、状态机实现，还是协程与异步IO，生成器都是Python程序员工具箱中不可或缺的利器。深入理解生成器，不仅能写出更高效、更Pythonic的代码，更能帮助你理解asyncio等现代异步框架的底层运行机制。