Python 生成器

def my_generator():
    yield 1
    yield 2
    yield 3
# 使用生成器
gen = my_generator()
print(next(gen))  # 输出: 1
print(next(gen))  # 输出: 2
print(next(gen))  # 输出: 3

# 类似于列表推导式，但使用圆括号
gen_expr = (x * x for x in range(5))
for value in gen_expr:
    print(value)  # 输出: 0, 1, 4, 9, 16

def generator_chain():
    """生成器委托"""
    yield from range(3)
    yield from ['a', 'b', 'c']
    yield from (x * 2 for x in range(2))

for item in generator_chain():
    print(item)  # 输出: 0,1,2,a,b,c,0,2

import sys

# 方式1：使用列表（占用大量内存）
def get_numbers_list(n):
    numbers = []
    for i in range(n):
        numbers.append(i)
    return numbers

# 方式2：使用生成器（几乎不占内存）
def get_numbers_generator(n):
    for i in range(n):
        yield i

# 内存占用对比
n = 1000000
list_memory = sys.getsizeof(get_numbers_list(n))
gen_memory = sys.getsizeof(get_numbers_generator(n))

print(f"列表占用内存: {list_memory} 字节")  # 约 8.5MB
print(f"生成器占用内存: {gen_memory} 字节")  # 仅 128 字节

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

# 只计算需要的值，不会无限循环
fib = fibonacci_infinite()
for i, num in enumerate(fib):
    if i >= 10:  # 只取前10个
        break
    print(num)  # 输出: 0, 1, 1, 2, 3, 5, 8, 13, 21, 34

def read_large_file(file_path):
    """逐行读取大文件，不加载到内存"""
    with open(file_path, 'r') as file:
        for line in file:
            yield line.strip()

# 处理GB级别的文件
for line in read_large_file('huge_file.txt'):
    process_line(line)  # 一次只处理一行

def stateful_generator():
    total = 0
    while True:
        value = yield total
        if value is None:
            break
        total += value

# 生成器保持内部状态
calculator = stateful_generator()
next(calculator)  # 启动生成器

print(calculator.send(5))   # 输出: 5
print(calculator.send(10))  # 输出: 15
print(calculator.send(3))   # 输出: 18

def read_data():
    """数据读取阶段"""
    for i in range(10):
        yield i

def filter_even(numbers):
    """过滤阶段"""
    for num in numbers:
        if num % 2 == 0:
            yield num

def square(numbers):
    """转换阶段"""
    for num in numbers:
        yield num * num

# 构建数据处理管道
pipeline = square(filter_even(read_data()))
for result in pipeline:
    print(result)  # 输出: 0, 4, 16, 36, 64

def paginate_data(total_items, page_size):
    """分页生成器"""
    page = 0
    while page * page_size < total_items:
        start = page * page_size
        end = start + page_size
        yield (start, end)
        page += 1

# 模拟分页处理数据
for start, end in paginate_data(100, 10):
    print(f"处理第 {start}-{end} 条数据")

import time
import random

def sensor_data_stream():
    """模拟传感器数据流"""
    while True:
        yield {
            'timestamp': time.time(),
            'temperature': random.uniform(20, 30),
            'humidity': random.uniform(40, 80)
        }
        time.sleep(1)  # 每秒产生一个数据点

# 处理实时数据
for data in sensor_data_stream():
    print(f"温度: {data['temperature']:.1f}°C")
    if data['temperature'] > 28:
        print("警告: 温度过高!")

# 计算平方和，不创建中间列表
sum_of_squares = sum(x*x for x in range(1000000))

# 文件行数统计
line_count = sum(1 for line in open('file.txt'))