你能否用Python案例解释协程的基本用法

wen python案例 2026-06-04 16:20:19 4

Python协程核心用法实战指南

目录导读

协程是什么？与线程有何区别？
环境准备与基础语法
同步爬虫 vs 协程爬虫
任务超时与异常处理
协程结合异步上下文管理器
常见问题与调试技巧
总结与延伸阅读

协程是什么？与线程有何区别？

在正式写代码之前,先解决一个核心疑问：

问：为什么不用线程，而用协程？
答：线程由操作系统调度，上下文切换开销大，且需处理锁、竞争条件，协程由程序自身调度，在单线程内实现“并发”，切换成本极低，适合I/O密集型任务（如网络请求、文件读写），Python 3.5+ 通过 async/await 语法原生支持协程。

类比：线程像多个人（高成本）完成多件事，协程像一个人（低成本）快速切换做不同事的片段。

环境准备与基础语法

你需要 Python 3.7+，并安装 aiohttp 和 asyncio（标准库）。

import asyncio
async def say_hello():
    print("Hello")
    await asyncio.sleep(1)  # 模拟IO等待
    print("World")
asyncio.run(say_hello())

关键点：

async def 定义协程函数
await 挂起当前协程，让出控制权
asyncio.run() 启动事件循环

案例一：同步爬虫 vs 协程爬虫

场景：请求三个网页，每个返回需要2秒。

同步版本

import time
import requests
def fetch(url):
    print(f"开始请求: {url}")
    response = requests.get(url)
    print(f"完成: {url}")
    return response.status_code
start = time.time()
for url in ["https://httpbin.org/delay/2"] * 3:
    fetch(url)
print(f"同步耗时: {time.time() - start:.2f}s")
# 输出: 同步耗时约6秒

协程版本

import asyncio
import aiohttp
async def fetch_async(session, url):
    print(f"开始请求: {url}")
    async with session.get(url) as response:
        print(f"完成: {url}")
        return response.status
async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_async(session, "https://httpbin.org/delay/2") for _ in range(3)]
        results = await asyncio.gather(*tasks)
        print("状态码:", results)
start = time.time()
asyncio.run(main())
print(f"协程耗时: {time.time() - start:.2f}s")
# 输出: 协程耗时约2秒

问：为什么能快3倍？
答：asyncio.gather() 并发创建3个任务，当遇到 await response 时，事件循环自动切换到另一个未完成的任务，总时间等于最慢任务的时间（2秒），而非3个任务之和。

案例二：任务超时与异常处理

实际应用中,某个请求可能卡死，使用 asyncio.wait_for 控制超时。

import asyncio
async def slow_operation():
    await asyncio.sleep(10)
    return "Done"
async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2)
        print(result)
    except asyncio.TimeoutError:
        print("任务超时，已取消")
asyncio.run(main())

当任务超时,asyncio.wait_for 会自动取消协程，避免资源泄漏。

案例三：协程结合异步上下文管理器

读写文件或数据库时,安全释放资源很重要。

import asyncio
class AsyncResource:
    async def __aenter__(self):
        print("获取资源")
        await asyncio.sleep(0.5)
        return self
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("释放资源")
        await asyncio.sleep(0.3)
async def use_resource():
    async with AsyncResource() as res:
        print("使用资源中")
        await asyncio.sleep(1)
asyncio.run(use_resource())

__aenter__ 和 __aexit__ 必须定义为异步，配合 async with 使用。

常见问题与调试技巧

问题1：协程函数被直接调用，未使用 await

async def foo(): pass
foo()  # 输出: <coroutine object foo at 0x...>，函数未执行

解决：要么 await foo()，要么 asyncio.run(foo())。

问题2：asyncio.run() 不能嵌套
在已有事件循环中调用 asyncio.run() 会报错，解决方案：内部使用 await 或 asyncio.create_task()。

问题3：阻塞代码混入协程（如 time.sleep）
这会阻塞整个事件循环，永远用 asyncio.sleep() 代替 time.sleep()。

调试技巧

设置 asyncio.get_event_loop().set_debug(True)
使用 nest_asyncio 库（需pip安装）允许在Jupyter中执行 asyncio.run()

总结与延伸阅读

核心关键词：事件循环、协程函数、await、asyncio.gather、异步上下文管理器。
应用场景：爬虫、Web框架（FastAPI/Starlette）、微服务网关、消息队列消费者。

问：掌握上述案例后，下一步学什么？
答：

学习 asyncio.Queue 实现生产者-消费者模式
理解 asyncio.Semaphore 控制并发数
研究 async generator（异步生成器）
对比线程池 + run_in_executor 的适用场景

现在你可以打开编辑器,将第一个爬虫案例跑起来，观察耗时差异，协程就像单线程里的“时间管理大师”，用最小的成本做最多的事。

标签： Python案例

本文地址： https://dfhcn.com/post/6.html

文章来源： wen