Python生成器案例如何编写生成器

wen python案例 2026-06-30 25

本文目录导读：

Python生成器案例如何编写生成器

基础生成器函数
生成器表达式
高级生成器模式
实用案例
性能比较示例
使用 yield from 委托

我来为您详细介绍Python生成器的编写方法,包含多个实用案例。

基础生成器函数

使用 `yield` 关键字

# 基础计数器生成器
def count_up_to(n):
    """生成从1到n的数字"""
    i = 1
    while i <= n:
        yield i
        i += 1
# 使用示例
counter = count_up_to(5)
for num in counter:
    print(num)  # 输出：1, 2, 3, 4, 5

斐波那契数列生成器

def fibonacci_generator(limit=None):
    """生成斐波那契数列"""
    a, b = 0, 1
    count = 0
    while True:
        if limit and count >= limit:
            return
        yield a
        a, b = b, a + b
        count += 1
# 使用示例
fib = fibonacci_generator(10)
print(list(fib))  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

生成器表达式

对比列表推导式

# 列表推导式（占用内存较大）
squares_list = [x**2 for x in range(1000000)]
# 生成器表达式（节省内存）
squares_generator = (x**2 for x in range(1000000))
print(type(squares_list))     # <class 'list'>
print(type(squares_generator)) # <class 'generator'>
# 使用生成器
for square in squares_generator:
    if square > 100:
        break
    print(square)  # 0, 1, 4, 9, ..., 100

实际应用示例

# 读取大文件
def read_large_file(file_path):
    """逐行读取大文件的生成器"""
    with open(file_path, 'r', encoding='utf-8') as file:
        for line in file:
            yield line.strip()
# 无限序列生成器
def infinite_cycle(elements):
    """无限循环生成元素"""
    while True:
        for element in elements:
            yield element
# 使用生成器表达式处理数据
data = ["Alice", "Bob", "Charlie"]
names = (name.upper() for name in data)
print(list(names))  # ['ALICE', 'BOB', 'CHARLIE']

高级生成器模式

管道（Pipeline）模式

def read_lines(file_path):
    """读取文件行"""
    with open(file_path, 'r') as f:
        for line in f:
            yield line.strip()
def filter_empty(lines):
    """过滤空行"""
    for line in lines:
        if line.strip():
            yield line
def transform_uppercase(lines):
    """转换为大写"""
    for line in lines:
        yield line.upper()
# 组合使用
def process_file(file_path):
    lines = read_lines(file_path)
    non_empty = filter_empty(lines)
    uppercase = transform_uppercase(non_empty)
    return uppercase
# 使用
for line in process_file('data.txt'):
    print(line)

协程生成器（双向通信）

def coroutine_average():
    """计算平均值的协程"""
    total = 0
    count = 0
    average = None
    while True:
        # 接收值
        new_value = yield average
        if new_value is None:
            break
        total += new_value
        count += 1
        average = total / count
# 使用示例
avg = coroutine_average()
next(avg)  # 启动生成器
print(avg.send(10))  # 10.0
print(avg.send(20))  # 15.0
print(avg.send(30))  # 20.0
avg.close()  # 关闭生成器

实用案例

分页数据生成器

def paginate_data(data, page_size=10):
    """数据分页生成器"""
    for i in range(0, len(data), page_size):
        yield data[i:i + page_size]
# 使用示例
users = [f"user_{i}" for i in range(100)]
pages = paginate_data(users, 20)
for page_num, page in enumerate(pages, 1):
    print(f"第 {page_num} 页: {page[:3]}...")  # 只显示前3个

实时数据监控生成器

import time
import random
def sensor_data_generator():
    """模拟传感器数据流"""
    while True:
        temperature = random.uniform(20, 30)
        humidity = random.uniform(40, 60)
        yield {"temperature": temperature, "humidity": humidity}
        time.sleep(0.1)  # 模拟采样间隔
# 使用生成器处理实时数据
def monitor_sensor(duration=5):
    """监控传感器数据一定时间"""
    sensor = sensor_data_generator()
    start_time = time.time()
    while time.time() - start_time < duration:
        data = next(sensor)
        if data["temperature"] > 28:
            print(f"警告：温度过高！{data['temperature']:.1f}°C")
        else:
            print(f"正常：{data['temperature']:.1f}°C, {data['humidity']:.1f}%")
# monitor_sensor(3)  # 监控3秒

性能比较示例

import sys
import time
def memory_comparison():
    """比较生成器和列表的内存使用"""
    # 列表方式
    start_time = time.time()
    squares_list = [x**2 for x in range(1000000)]
    list_size = sys.getsizeof(squares_list)
    list_time = time.time() - start_time
    # 生成器方式
    start_time = time.time()
    squares_gen = (x**2 for x in range(1000000))
    gen_size = sys.getsizeof(squares_gen)
    gen_time = time.time() - start_time
    print(f"列表大小: {list_size} 字节, 创建时间: {list_time:.4f}秒")
    print(f"生成器大小: {gen_size} 字节, 创建时间: {gen_time:.4f}秒")
    print(f"内存节省: {list_size / gen_size:.0f} 倍")
# memory_comparison()

使用 `yield from` 委托

def chain_generators(*iterables):
    """连接多个可迭代对象"""
    for iterable in iterables:
        yield from iterable
# 简化版本
def simple_chain(*iterables):
    """更简洁的连接方式"""
    for iterable in iterables:
        yield from iterable
# 使用示例
list1 = [1, 2, 3]
list2 = [4, 5, 6]
list3 = [7, 8, 9]
chained = chain_generators(list1, list2, list3)
print(list(chained))  # [1, 2, 3, 4, 5, 6, 7, 8, 9]