【Python知识详解】并发编程：asyncio 异步编程入门

前言

在现代编程中，处理 I/O 密集型任务（如网络请求、文件读写）时，同步代码往往会阻塞等待，导致程序效率低下。Python 的 asyncio 模块提供了异步编程范式，让你在单线程内高效处理大量 I/O 操作。本文将系统讲解 asyncio 的核心概念和使用方法。

一、为什么需要异步编程？

1.1 同步 vs 异步

# 同步：等待每个请求完成
import requests
​
def sync_fetch_all(urls):
   results = []
   for url in urls:
       response = requests.get(url)  # 阻塞等待
       results.append(response.json())
   return results
​
# 异步：并发处理请求
import asyncio
import aiohttp
​
async def async_fetch_all(urls):
   async with aiohttp.ClientSession() as session:
       tasks = [fetch(session, url) for url in urls]
       return await asyncio.gather(*tasks)
​
async def fetch(session, url):
   async with session.get(url) as response:
       return await response.json()

1.2 性能对比

同步处理 10 个请求（每个 1 秒）：
总耗时 ≈ 10 秒
​
异步处理 10 个请求（每个 1 秒）：
总耗时 ≈ 1 秒

二、asyncio 基础

2.1 协程函数

import asyncio
​
# async def 定义的函数是协程函数
async def hello():
   print("Hello")
   await asyncio.sleep(1)
   print("World")
​
# 调用协程函数会返回一个协程对象
coro = hello()
print(type(coro))  # <class 'coroutine'>
​
# 运行协程
asyncio.run(coro)

2.2 await 关键字

import asyncio
​
async def say_after(delay, what):
   await asyncio.sleep(delay)
   print(what)
​
async def main():
   # 顺序执行（耗时 3 秒）
   await say_after(1, "hello")
   await say_after(2, "world")
​
   # 并发执行（耗时 2 秒）
   await asyncio.gather(
       say_after(1, "hello"),
       say_after(2, "world")
  )
​
asyncio.run(main())

2.3 asyncio.run()

# Python 3.7+ 推荐用法
async def main():
   # 协程代码
   pass
​
asyncio.run(main())  # 入口点

三、Task 和 Future

3.1 创建 Task

import asyncio
​
async def myCoroutine():
   print("Task 执行中")
   await asyncio.sleep(1)
   return "结果"
​
async def main():
   # 创建 Task（调度协程执行）
   task = asyncio.create_task(myCoroutine())
​
   # Task 对象可等待
   result = await task
   print(f"结果：{result}")
​
asyncio.run(main())

3.2 并发执行多个 Task

import asyncio
​
async def fetch(url):
   await asyncio.sleep(1)  # 模拟 I/O
   return f"Fetched {url}"
​
async def main():
   # 创建多个 Task
   urls = ["url1", "url2", "url3", "url4", "url5"]
   tasks = [asyncio.create_task(fetch(url)) for url in urls]
​
   # 等待所有 Task 完成
   results = await asyncio.gather(*tasks)
   print(results)
​
asyncio.run(main())  # 耗时约 1 秒（并发执行）

3.3 asyncio.sleep vs time.sleep

# time.sleep 是阻塞的，会阻塞整个线程
import time
​
def sync_sleep():
   time.sleep(1)  # 阻塞 1 秒，期间什么都做不了
​
# asyncio.sleep 不会阻塞事件循环
import asyncio
​
async def async_sleep():
   await asyncio.sleep(1)  # 暂停协程，但不阻塞其他协程

四、异步上下文管理器

4.1 async with

import asyncio
​
class AsyncResource:
   async def __aenter__(self):
       print("获取资源")
       return self
​
   async def __aexit__(self, exc_type, exc_val, exc_tb):
       print("释放资源")
       return False  # 不抑制异常
​
async def main():
   async with AsyncResource() as resource:
       print("使用资源")
​
asyncio.run(main())
​
# 输出：
# 获取资源
# 使用资源
# 释放资源

4.2 aiohttp 示例

import asyncio
import aiohttp
​
async def fetch(session, url):
   async with session.get(url) as response:
       return await response.text()
​
async def main():
   urls = [
       "https://httpbin.org/get",
       "https://httpbin.org/get",
       "https://httpbin.org/get",
  ]
​
   async with aiohttp.ClientSession() as session:
       tasks = [fetch(session, url) for url in urls]
       responses = await asyncio.gather(*tasks)
       for i, resp in enumerate(responses):
           print(f"响应 {i}: {len(resp)} 字节")
​
asyncio.run(main())

五、异步迭代器

5.1 async for

import asyncio
​
class AsyncCounter:
   def __init__(self, max):
       self.max = max
       self.current = 0
​
   def __aiter__(self):
       return self
​
   async def __anext__(self):
       if self.current >= self.max:
           raise StopAsyncIteration
       value = self.current
       self.current += 1
       await asyncio.sleep(0.1)
       return value
​
async def main():
   async for i in AsyncCounter(5):
       print(i)
​
asyncio.run(main())

5.2 异步生成器

import asyncio
​
async def async_range(start, end, step=1):
   """异步生成器"""
   current = start
   while current < end:
       yield current
       await asyncio.sleep(0.1)
       current += step
​
async def main():
   async for i in async_range(0, 5):
       print(i)
​
asyncio.run(main())

六、实战案例

6.1 异步文件下载器

"""
异步文件下载器
"""
​
import asyncio
import aiohttp
from pathlib import Path
​
class AsyncDownloader:
   """异步下载器"""
​
   def __init__(self, max_concurrent=5):
       self.max_concurrent = max_concurrent
       self.semaphore = None
       self.results = []
​
   async def download(self, session, url, save_path):
       """下载单个文件"""
       async with self.semaphore:  # 限制并发数
           try:
               async with session.get(url) as response:
                   if response.status == 200:
                       content = await response.read()
                       Path(save_path).write_bytes(content)
                       return {"url": url, "status": "success", "size": len(content)}
                   else:
                       return {"url": url, "status": "error", "code": response.status}
           except Exception as e:
               return {"url": url, "status": "exception", "error": str(e)}
​
   async def download_all(self, urls):
       """下载多个文件"""
       self.semaphore = asyncio.Semaphore(self.max_concurrent)
       self.results = []
​
       async with aiohttp.ClientSession() as session:
           tasks = [
               self.download(session, url, f"downloads/{i}_{url.split('/')[-1]}")
               for i, url in enumerate(urls)
          ]
           self.results = await asyncio.gather(*tasks)
​
       return self.results
​
   def report(self):
       """生成报告"""
       total = len(self.results)
       success = sum(1 for r in self.results if r["status"] == "success")
       errors = total – success
​
       print("\\n📊 下载报告")
       print("=" * 50)
       print(f"总数：{total}")
       print(f"成功：{success} ({success/total*100:.1f}%)")
       print(f"失败：{errors} ({errors/total*100:.1f}%)")
​
       if success > 0:
           total_size = sum(r.get("size", 0) for r in self.results if r["status"] == "success")
           print(f"总大小：{total_size / 1024:.2f} KB")
​
​
async def main():
   # 创建下载目录
   Path("downloads").mkdir(exist_ok=True)
​
   # 测试 URLs
   urls = [
       "https://httpbin.org/bytes/1024",   # 1KB
       "https://httpbin.org/bytes/2048",   # 2KB
       "https://httpbin.org/bytes/10240",  # 10KB
  ]
​
   print("📥 异步下载器演示")
   print("=" * 50)
​
   downloader = AsyncDownloader(max_concurrent=3)
   await downloader.download_all(urls)
   downloader.report()
​
​
if __name__ == "__main__":
   asyncio.run(main())

6.2 异步 Web 服务器

"""
异步 Web 服务器示例
使用 aiohttp
"""
​
import asyncio
from aiohttp import web
​
# 模拟数据库
users_db = {
   "1": {"name": "张三", "age": 25},
   "2": {"name": "李四", "age": 30},
   "3": {"name": "王五", "age": 28},
}
​
# 路由处理函数
async def get_users(request):
   """获取所有用户"""
   return web.json_response(list(users_db.values()))
​
async def get_user(request):
   """获取单个用户"""
   user_id = request.match_info.get("id")
​
   if user_id not in users_db:
       return web.json_response({"error": "用户不存在"}, status=404)
​
   return web.json_response(users_db[user_id])
​
async def create_user(request):
   """创建用户"""
   try:
       data = await request.json()
   except:
       return web.json_response({"error": "无效的 JSON"}, status=400)
​
   if "name" not in data or "age" not in data:
       return web.json_response({"error": "缺少必要字段"}, status=400)
​
   user_id = str(max(int(k) for k in users_db.keys()) + 1)
   users_db[user_id] = {"name": data["name"], "age": data["age"]}
​
   return web.json_response({"id": user_id, **users_db[user_id]}, status=201)
​
async def delete_user(request):
   """删除用户"""
   user_id = request.match_info.get("id")
​
   if user_id not in users_db:
       return web.json_response({"error": "用户不存在"}, status=404)
​
   del users_db[user_id]
   return web.json_response({"message": "删除成功"})
​
# 应用设置
app = web.Application()
​
# 添加路由
app.router.add_get("/users", get_users)
app.router.add_get("/users/{id}", get_user)
app.router.add_post("/users", create_user)
app.router.add_delete("/users/{id}", delete_user)
​
# 健康检查
async def health(request):
   return web.json_response({"status": "ok"})
​
app.router.add_get("/health", health)
​
if __name__ == "__main__":
   print("🌐 异步 Web 服务器")
   print("=" * 50)
   print("启动服务器：http://localhost:8080")
   print("\\n可用接口：")
   print(" GET   /users       – 获取所有用户")
   print(" GET   /users/{id}   – 获取单个用户")
   print(" POST   /users       – 创建用户")
   print(" DELETE /users/{id}   – 删除用户")
   print(" GET   /health       – 健康检查")
   print("\\n按 Ctrl+C 停止服务器")
   print("=" * 50)
​
   web.run_app(app, host="localhost", port=8080)

6.3 异步爬虫

"""
异步爬虫示例
"""
​
import asyncio
import aiohttp
from bs4 import BeautifulSoup
import re
​
class AsyncSpider:
   """异步爬虫"""
​
   def __init__(self, max_concurrent=10):
       self.max_concurrent = max_concurrent
       self.semaphore = None
       self.results = []
​
   async def fetch_page(self, session, url):
       """获取单个页面"""
       async with self.semaphore:
           try:
               async with session.get(url, timeout=10) as response:
                   html = await response.text()
                   return self.parse_page(url, html)
           except Exception as e:
               return {"url": url, "error": str(e)}
​
   def parse_page(self, url, html):
       """解析页面，提取链接"""
       soup = BeautifulSoup(html, "html.parser")
​
       # 提取标题
       title = soup.find("title")
       title_text = title.get_text().strip() if title else "无标题"
​
       # 提取所有链接
       links = []
       for a in soup.find_all("a", href=True):
           href = a["href"]
           if href.startswith("http"):
               links.append(href)
​
       return {
           "url": url,
           "title": title_text,
           "links_count": len(links),
           "links": links[:5],  # 只保留前5个
      }
​
   async def crawl(self, urls):
       """爬取多个页面"""
       self.semaphore = asyncio.Semaphore(self.max_concurrent)
​
       async with aiohttp.ClientSession() as session:
           tasks = [self.fetch_page(session, url) for url in urls]
           self.results = await asyncio.gather(*tasks)
​
       return self.results
​
   def report(self):
       """生成报告"""
       print("\\n📊 爬取报告")
       print("=" * 60)
​
       for result in self.results:
           if "error" in result:
               print(f"❌ {result['url']}: {result['error']}")
           else:
               print(f"✅ {result['title']}")
               print(f"   URL: {result['url']}")
               print(f"   链接数: {result['links_count']}")
​
               if result['links']:
                   print(f"   示例链接: {result['links'][0][:50]}…")
               print()
​
​
async def main():
   print("🕷️ 异步爬虫演示")
   print("=" * 50)
​
   # 测试 URLs
   urls = [
       "https://httpbin.org/html",
       "https://httpbin.org/links/10",
       "https://httpbin.org/html",
  ]
​
   spider = AsyncSpider(max_concurrent=5)
   await spider.crawl(urls)
   spider.report()
​
​
if __name__ == "__main__":
   asyncio.run(main())

七、异步 vs 多线程 vs 多进程

7.1 选择指南

7.2 对比示例

# 多线程（适合 I/O）
from concurrent.futures import ThreadPoolExecutor
import requests
​
def fetch_url(url):
   return requests.get(url).json()
​
with ThreadPoolExecutor(max_workers=10) as executor:
   results = list(executor.map(fetch_url, urls))
​
# 多进程（适合 CPU）
from concurrent.futures import ProcessPoolExecutor
​
def cpu_task(n):
   return sum(i * i for i in range(n))
​
with ProcessPoolExecutor(max_workers=4) as executor:
   results = list(executor.map(cpu_task, [1000000] * 8))
​
# 异步（适合 I/O，等待期间可处理其他任务）
import asyncio
import aiohttp
​
async def fetch_all(urls):
   async with aiohttp.ClientSession() as session:
       tasks = [fetch(session, url) for url in urls]
       return await asyncio.gather(*tasks)

八、常见面试题

面试题1：async/await 的作用？

# async def 定义协程函数
# await 等待协程/任务完成
# 不能在普通函数中使用 await

面试题2：asyncio.gather vs asyncio.create_task？

# gather – 并发执行多个协程，返回结果列表
results = await asyncio.gather(task1, task2, task3)
​
# create_task – 创建任务调度执行
task1 = asyncio.create_task(coro1())
task2 = asyncio.create_task(coro2())
# 任务已调度，可以先做其他事
result1 = await task1
result2 = await task2

面试题3：什么是事件循环？

# 事件循环驱动协程执行
# asyncio.run() 创建并运行事件循环
# 事件循环不断检查就绪的协程并执行

总结

相关文章：

• 【Python知识详解】网络编程——socket 通信

• 【Python知识详解】数据库操作——SQLite/MySQL

如果觉得文章有帮助，欢迎关注、点赞、收藏！