Python协程案例如何实现异步协程

wen python案例 2026-06-30 30

Python协程案例：如何实现异步协程？从入门到实战详解

📚 目录导读

什么是Python协程？为什么需要异步编程？
核心概念：async/await、事件循环、Future与Task
案例1：基础异步任务——模拟网络请求
案例2：并发执行多个协程——使用gather与as_completed
案例3：带超时的协程控制——asyncio.wait_for
案例4：异步队列处理——生产者消费者模型
常见错误与调试技巧
QA问答：协程与多线程的区别？如何避免阻塞？

什么是Python协程？为什么需要异步编程？

在传统的同步编程中，当程序遇到I/O操作（如文件读写、网络请求）时会阻塞，等待操作完成后再继续执行，如果同时有100个网络请求，同步代码会依次等待，耗时巨大。

Python协程案例如何实现异步协程

协程（Coroutine） 是一种更轻量的并发方案，它通过单线程内的协作式多任务，允许一个任务在执行到I/O时主动“挂起”，让出CPU给其他任务，等I/O完成后再恢复，Python的asyncio库提供了完整的异步框架，核心是事件循环（Event Loop）。

关键类比：协程就像“只会暂时睡觉的助手”，而多线程是“多个吵闹的同事”，协程切换开销极低（微秒级），且没有线程安全问题。

核心概念：async/await、事件循环、Future与Task

概念	说明	示例
`async def`	定义协程函数	`async def fetch_data():`
`await`	挂起协程，等待另一个协程完成	`data = await coroutine`
`事件循环`	调度协程运行的核心引擎	`asyncio.run(main())`
`Task`	将协程包装为可独立调度任务	`task = asyncio.create_task(coro)`
`Future`	表示异步操作的最终结果（通常不直接使用）	底层由Task继承实现

事件循环的工作流程：

将协程包装成Task。
轮询所有Task，若有就绪的则执行。
遇到await时挂起当前Task，切换到下一个就绪Task。
当I/O完成时，恢复挂起的Task。

案例1：基础异步任务——模拟网络请求

import asyncio
import time
async def fetch_url(url, delay):
    print(f"开始请求 {url}")
    await asyncio.sleep(delay)  # 模拟网络I/O
    print(f"完成请求 {url}")
    return f"数据来自 {url}"
async def main():
    start = time.time()
    # 顺序执行两个协程（如同步）
    result1 = await fetch_url("http://example.com/1", 2)
    result2 = await fetch_url("http://example.com/2", 3)
    print(f"总耗时: {time.time() - start:.2f}秒")
    print(result1, result2)
asyncio.run(main())

输出（耗时约5秒）：

开始请求 http://example.com/1  
完成请求 http://example.com/1  
开始请求 http://example.com/2  
完成请求 http://example.com/2  
总耗时: 5.01秒

注意：直接await协程是顺序执行，没有并发,想要并发需要创建Task。

案例2：并发执行多个协程——gather与as_completed

方法A：`asyncio.gather` —— 等待所有结果

async def main_concurrent():
    start = time.time()
    # 并发执行3个任务
    tasks = [
        fetch_url("http://site.com/1", 2),
        fetch_url("http://site.com/2", 3),
        fetch_url("http://site.com/3", 1)
    ]
    results = await asyncio.gather(*tasks)  # 返回结果列表
    print(f"并发总耗时: {time.time() - start:.2f}秒")
    print(results)
asyncio.run(main_concurrent())

输出（耗时约3秒，取最大延迟）：

开始请求 http://site.com/1  
开始请求 http://site.com/2  
开始请求 http://site.com/3  
完成请求 http://site.com/3  
完成请求 http://site.com/1  
完成请求 http://site.com/2  
并发总耗时: 3.01秒

方法B：`asyncio.as_completed` —— 按完成顺序处理

async def process_as_completed():
    tasks = [
        fetch_url("A", 3),
        fetch_url("B", 1),
        fetch_url("C", 2)
    ]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"先完成的: {result}")
asyncio.run(process_as_completed())

输出顺序：B -> C -> A

案例3：带超时的协程控制——asyncio.wait_for

当某个协程可能阻塞过久时,可以设置超时：

async def slow_operation():
    await asyncio.sleep(10)
    return "慢操作完成"
async def timeout_demo():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(result)
    except asyncio.TimeoutError:
        print("操作超时！")
asyncio.run(timeout_demo())

输出：操作超时！

案例4：异步队列处理——生产者消费者模型

适用于爬虫、日志处理等场景：

import asyncio
from random import randint
async def producer(queue, n):
    for i in range(n):
        item = f"任务{i+1}"
        await queue.put(item)
        print(f"生产: {item}")
        await asyncio.sleep(randint(1,2))  # 模拟生产延迟
    await queue.put(None)  # 终止信号
async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        print(f"{name} 消费: {item}")
        await asyncio.sleep(randint(1,3))  # 模拟处理延迟
        queue.task_done()
async def main_queue():
    queue = asyncio.Queue(maxsize=3)
    # 创建多个消费者
    consumers = [consumer(queue, f"工人{i}") for i in range(2)]
    await asyncio.gather(
        producer(queue, 5),
        *consumers
    )
asyncio.run(main_queue())

常见错误与调试技巧

忘记await：直接调用协程会产生警告RuntimeWarning: coroutine was never awaited。
✅ 正确：result = await my_coro()
阻塞事件循环：在协程内部使用time.sleep()会阻塞整个循环。
✅ 替代：await asyncio.sleep(秒数)
多个事件循环：在Jupyter Notebook中可能遇到，使用nest_asyncio.apply()解决。
调试工具：设置asyncio.get_event_loop().set_debug(True)可查看任务切换详情。

❓ QA问答：协程常见疑问

Q1：协程与多线程的主要区别？

协程：单线程，协作式切换，适合I/O密集型任务（网络、文件），切换开销极低，无锁竞争。
多线程：多线程，抢占式切换，适合CPU密集型或阻塞性I/O，需处理锁、死锁问题。

Q2：协程能提高CPU密集型任务的性能吗？
不能，协程本质上仍是单线程，无法利用多核，CPU密集任务应使用multiprocessing或concurrent.futures.ProcessPoolExecutor。

Q3：如何避免协程中意外阻塞？

所有需要阻塞的库必须使用异步版本（如aiohttp替代requests，asyncpg替代psycopg2）。
如果必须使用同步库，用asyncio.to_thread()将阻塞操作扔到线程池。

result = await asyncio.to_thread(sync_function, arg1)

Q4：asyncio.run()和loop.run_until_complete()哪个更好？
asyncio.run()是Python 3.7+的推荐方式，自动创建/关闭事件循环，在复杂应用（如长期服务）中可手动管理循环。

总结与最佳实践

何时使用协程：大量I/O操作（>100个并发），如Web爬虫、API聚合、聊天服务器。
何时避免：CPU密集型任务、低延迟要求（协程的调度延迟在微妙级，但仍可接受）。
性能建议：使用asyncio.Semaphore控制并发数，避免一次性创建过多任务。

延伸资源：

官方文档：asyncio模块详解
第三方库：aiohttp（HTTP客户端/服务端）、aiomysql（数据库）

通过以上案例，你应该已经掌握Python异步协程的核心用法，从简单的await到复杂的队列处理，逐步实践，你会发现异步编程能让I/O密集型任务效率提升数倍！

Python协程案例如何实现异步协程