Mastering Python Async: From Generators to Async/Await and Real-World Use Cases

1. Generator Basics: The Power of yield

Generators are special iterators that produce values lazily using the yield statement, which saves memory compared with building full lists.

def simple_generator():
    print("开始执行")
    yield 1
    print("继续执行")
    yield 2
    print("结束执行")
    yield 3

Memory comparison shows a generator uses virtually no memory while a list of one million integers occupies several megabytes.

import sys

def get_numbers_list(n):
    numbers = []
    for i in range(n):
        numbers.append(i)
    return numbers

def get_numbers_gen(n):
    for i in range(n):
        yield i

n = 1_000_000
list_memory = sys.getsizeof(get_numbers_list(n))
gen_memory = sys.getsizeof(get_numbers_gen(n))
print(f"列表内存占用: {list_memory/1024/1024:.2f} MB")
print(f"生成器内存占用: {gen_memory/1024/1024:.2f} MB")

2. Generator Expressions

Generator expressions provide a concise way to create generators without a full function definition.

squares_gen = (x * x for x in range(1_000_000))
print(f"生成器大小: {sys.getsizeof(squares_gen)} bytes")

3. yield as an Expression (Two‑Way Communication)

Since Python 2.5, yield can receive a value sent from the caller, enabling simple coroutines.

def interactive_generator():
    print("生成器启动")
    received = yield "请发送数据"
    print(f"收到: {received} -> 请发送更多数据")
    received = yield "请发送更多数据"
    print(f"收到: {received} -> 完成")
    yield "完成"

gen = interactive_generator()
print(next(gen))               # 启动，输出 "请发送数据"
print(gen.send("你好"))        # 发送第一条数据
print(gen.send("世界"))        # 发送第二条数据

4. Asyncio and Native Coroutines

Python 3.4 introduced the asyncio module; Python 3.5 added async / await syntax for true native coroutines.

import asyncio, time

async def say_after(delay, message):
    await asyncio.sleep(delay)
    print(message)
    return f"完成: {message}"

async def main():
    print(f"开始时间: {time.strftime('%X')}")
    result1 = await say_after(2, "你好")
    result2 = await say_after(1, "世界")
    print(f"结束时间: {time.strftime('%X')}")
    print(f"结果: {result1}, {result2}")

asyncio.run(main())

5. Concurrent Execution with asyncio.gather

Multiple coroutines can run concurrently using asyncio.create_task and asyncio.gather.

async def concurrent_main():
    print(f"开始时间: {time.strftime('%X')}")
    task1 = asyncio.create_task(say_after(2, "你好"))
    task2 = asyncio.create_task(say_after(1, "世界"))
    results = await asyncio.gather(task1, task2)
    print(f"结束时间: {time.strftime('%X')}")
    print(f"所有结果: {results}")

asyncio.run(concurrent_main())

6. Real‑World Async Examples

6.1 Asynchronous HTTP Requests

import aiohttp, asyncio

async def fetch_url(session, url):
    try:
        async with session.get(url, timeout=10) as response:
            content = await response.text()
            return f"{url}: {len(content)} bytes"
    except Exception as e:
        return f"{url}: 错误 - {e}"

async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/3",
        "https://invalid-url-test.com",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, u) for u in urls]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(r)

asyncio.run(main())

6.2 Asynchronous File I/O with aiofiles

import aiofiles, asyncio

async def async_file_operations():
    async with aiofiles.open('test.txt', 'w') as f:
        await f.write('Hello, async world!
')
        await f.write('This is async file IO.
')
    async with aiofiles.open('test.txt', 'r') as f:
        content = await f.read()
        print('文件内容:')
        print(content)
    async with aiofiles.open('test.txt', 'r') as f:
        async for line in f:
            print(f"行: {line.strip()}")

asyncio.run(async_file_operations())

6.3 Producer‑Consumer Pattern

import asyncio, random

async def producer(queue, pid):
    for i in range(5):
        item = f"产品-{pid}-{i}"
        await asyncio.sleep(random.uniform(0.1, 0.5))
        await queue.put(item)
        print(f"生产者 {pid} 生产了: {item}")
    await queue.put(None)  # 结束信号

async def consumer(queue, cid):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)
            break
        print(f"消费者 {cid} 消费了: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.8))

async def main():
    queue = asyncio.Queue(maxsize=10)
    producers = [producer(queue, i) for i in range(3)]
    consumers = [consumer(queue, i) for i in range(2)]
    await asyncio.gather(*producers, *consumers)

asyncio.run(main())

7. Performance Comparison: Synchronous vs Asynchronous I/O

The script measures execution time for fetching the same URL ten times using requests (synchronous) and aiohttp (asynchronous).

import time, requests, aiohttp, asyncio

def sync_fetch(urls):
    results = []
    for url in urls:
        r = requests.get(url)
        results.append(f"{url}: {len(r.text)} bytes")
    return results

async def async_fetch(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, u) for u in urls]
        return await asyncio.gather(*tasks)

urls = ["https://httpbin.org/delay/1"] * 10
start = time.time(); sync_fetch(urls); sync_time = time.time() - start
print(f"同步耗时: {sync_time:.2f} 秒")
start = time.time(); asyncio.run(async_fetch(urls)); async_time = time.time() - start
print(f"异步耗时: {async_time:.2f} 秒")
print(f"性能提升: {sync_time/async_time:.1f} 倍")

8. Error Handling and Debugging

8.1 Asynchronous Exception Handling

async def risky_operation():
    await asyncio.sleep(0.1)
    if random.random() < 0.3:
        raise ValueError("随机失败")
    return "成功"

async def robust_main():
    tasks = [asyncio.create_task(risky_operation()) for _ in range(10)]
    results = []
    for t in tasks:
        try:
            results.append(await t)
        except ValueError as e:
            print(f"任务失败: {e}")
            results.append("失败")
    print(f"完成结果: {results}")

asyncio.run(robust_main())

8.2 Debugging with Logging

import logging, asyncio
logging.basicConfig(level=logging.DEBUG)

async def debug_coroutine():
    logging.debug("协程开始")
    await asyncio.sleep(0.1)
    logging.debug("第一步完成")
    await asyncio.sleep(0.1)
    logging.debug("第二步完成")
    return "完成"

asyncio.run(debug_coroutine(), debug=True)

9. Best Practices & Common Pitfalls

Prefer async/await over legacy generator‑based coroutines.

Control concurrency with semaphores or queue size limits.

Always handle exceptions inside coroutines.

Set appropriate timeouts for I/O operations.

Avoid blocking calls like time.sleep inside async code; use await asyncio.sleep instead.

Typical mistakes include forgetting await on coroutine calls or using blocking functions inside async functions, which can stall the event loop.

10. Evolution Timeline

版本

特性

意义

Python 2.2

生成器（yield）

引入惰性计算

Python 2.5

yield 表达式

支持双向通信

Python 3.3

yield from

生成器委托

Python 3.4

asyncio 模块

异步编程框架

Python 3.5

async/await

原生协程支持

Python 3.7

asyncio.run()

简化异步入口

Choosing the right tool depends on the workload: I/O‑bound tasks benefit from async/await, CPU‑bound tasks from multiprocessing, simple lazy calculations from generators, and complex async workflows from the asyncio ecosystem.