生成器与异步
处理大数据流时，把全部数据塞进内存往往不现实。生成器让我们按需"惰性"产出；处理大量 I/O 时，异步让我们在等的过程中去做别的事。这两件事是 Python 高效编程的支柱。
1. 生成器函数：yield
包含 yield 的函数不再返回普通值，而是返回一个生成器对象。每次 next() 都会从上次 yield 处继续执行。
def counter(n):
i = 0
while i < n:
yield i
i += 1
# 手动驱动
g = counter(3)
print(next(g)) # 0
print(next(g)) # 1
print(next(g)) # 2
try:
print(next(g))
except StopIteration:
print("迭代结束")counter(10**18) 瞬间返回，不会生成 10^18 个数。值是"被问到时"才产生的。处理大文件、无限序列、流时这是核心优势。
2. 生成器表达式
类似列表推导式，但用圆括号，结果是生成器而不是列表。
# 列表推导式：立刻在内存中建好列表
squares_list = [x * x for x in range(5)]
print("list:", squares_list, "size:", len(squares_list))
# 生成器表达式：惰性
squares_gen = (x * x for x in range(5))
print("gen:", squares_gen, "type:", type(squares_gen).__name__)
# 用 sum() 消费
total = sum(x * x for x in range(10))
print("1²+2²+...+9² =", total)传给 sum()、min()、max()、any()、all() 时，可以省掉外层的圆括号：sum(x*x for x in range(10))。能懒则懒——但如果需要重复迭代或索引访问，必须物化成 list。
3. itertools 常用工具
itertools 提供了一系列组合迭代器的"积木"。用熟这些能省掉大量手写循环。
from itertools import count, islice, cycle, chain, groupby
# count(start, step): 无限计数
print("count(10, 2) 前 5 项:", list(islice(count(10, 2), 5)))
# cycle: 无限循环
print("cycle('AB') 前 6 项:", list(islice(cycle("AB"), 6)))
# chain: 把多个迭代器首尾相接
print("chain:", list(chain([1, 2], (3, 4), "ab")))
# groupby: 连续相同 key 聚成一组
words = ["apple", "ant", "banana", "blue", "berry", "cherry"]
for key, group in groupby(sorted(words), key=lambda w: w[0]):
print(key, "->", list(group))groupby 只会把连续相同的 key 聚成一组。上面我们先 sorted(words) 才会出现 a-b-c 三个分组；直接 groupby 原列表会得到错乱结果。
4. 异步函数：async def / await
async def 定义协程函数，调用它返回一个协程对象，必须用 await 驱动或交给事件循环。
import asyncio
async def fetch(name, delay):
print(f" [{name}] 开始")
await asyncio.sleep(delay) # 非阻塞地等
print(f" [{name}] 结束")
return f"{name}-结果"
async def main():
result = await fetch("A", 0.1)
print("拿到:", result)
asyncio.run(main())time.sleep(0.1) 会阻塞整个线程。asyncio.sleep(0.1) 只是把当前协程挂起，让事件循环去跑别的协程——这就是"非阻塞"的核心。
5. 并发运行：asyncio.gather
asyncio.gather 把多个协程打包，并发地跑（在 I/O 等待期间切换）：
import asyncio, time
async def task(name, secs):
print(f" [{name}] 启动，将等 {secs}s")
await asyncio.sleep(secs)
print(f" [{name}] 完成")
return f"{name}={secs}"
async def main():
start = time.perf_counter()
# 三个任务串行需要 1+2+3=6s；并发只需 3s
results = await asyncio.gather(
task("A", 1.0),
task("B", 2.0),
task("C", 3.0),
)
print("结果:", results)
print(f"总耗时: {time.perf_counter() - start:.2f}s")
asyncio.run(main())6. asyncio.create_task 与超时
gather 适合"全都要"；create_task 适合"启动后不管"；wait_for 提供超时控制。
import asyncio
async def slow():
await asyncio.sleep(10)
return "完成"
async def main():
# 启动但不等待
task = asyncio.create_task(slow())
# 等 0.2 秒，如果还没好就取消
try:
result = await asyncio.wait_for(task, timeout=0.2)
print(result)
except asyncio.TimeoutError:
print("超时，取消任务")
task.cancel()
try:
await task
except asyncio.CancelledError:
print("任务已取消")
asyncio.run(main())7. 异步生成器（async generator）
Python 把 yield 和 async 结合了起来，可以用 async for 消费。
import asyncio
async def ticker(n, interval):
for i in range(n):
await asyncio.sleep(interval)
yield i
async def main():
async for x in ticker(3, 0.01):
print("tick", x)
asyncio.run(main())🎯 练习
写一个并发示例：3 个"模拟请求"任务，分别耗时 1/2/3 秒。用 asyncio.gather 并发跑，统计总耗时并打印每个任务的返回值（包含名字和耗时）。
import asyncio, time
async def fetch(name, secs):
# 模拟一个网络请求：等待 secs 秒后返回结果
pass
async def main():
start = time.perf_counter()
# 在这里并发跑 fetch("A", 1.0) / fetch("B", 2.0) / fetch("C", 3.0)
# 期望总耗时 ~3s 而非 6s
pass
asyncio.run(main())fetch内部用await asyncio.sleep(secs)模拟 I/O 等待。- 用
await asyncio.gather(...)同时跑三个。 - 内部可以记录开始 / 结束时间，返回
(name, elapsed)。
小结
- ✅ 含
yield的函数返回生成器，next()一次走一步 - ✅ 生成器表达式
(x for x in xs)惰性求值，节省内存 - ✅
itertools.count / cycle / chain / islice / groupby是迭代器工具箱 - ✅
async def定义协程；await等待；asyncio.run()启动事件循环 - ✅
asyncio.gather并发跑多个协程；wait_for提供超时；create_task启动后台任务 - ✅ 异步生成器用
async for消费
下一章 上下文管理器：用 with 把"获取资源 / 释放资源"封装得安全又优雅。