三亩地 三亩地SAN MU DI · CODE DIARY
交流投稿
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天，欢迎挑你感兴趣的翻一翻。

首页/学习日记/日记详情

Python异步编程核心概念与实战技巧

发布时间：2026/8/10 5:54:52 三亩地 编程学习日记
Python异步编程核心概念与实战技巧

1. 为什么需要异步编程？

在传统的同步编程模型中，代码按照顺序执行，当遇到I/O操作（如网络请求、文件读写）时，整个程序会被阻塞，直到操作完成。这种模式在单线程环境下效率极低，因为CPU大部分时间都在等待I/O操作完成。

举个例子，假设我们要从三个不同的API获取数据：

import requests def fetch_data_sync(): data1 = requests.get('https://api1.example.com').json() # 阻塞 data2 = requests.get('https://api2.example.com').json() # 阻塞 data3 = requests.get('https://api3.example.com').json() # 阻塞 return [data1, data2, data3]

这段代码的总执行时间至少是三个请求响应时间的总和。而异步编程可以让我们在等待一个请求响应时，去处理其他任务。

注意：Python的requests库是同步的，异步编程需要使用专门的异步HTTP客户端如aiohttp

2. Python异步编程核心概念

2.1 事件循环(Event Loop)

事件循环是异步编程的核心引擎，它负责调度和执行协程。你可以把它想象成一个无限循环，不断检查哪些协程可以继续执行，哪些需要等待I/O。

import asyncio async def main(): print('Hello') await asyncio.sleep(1) print('World') # 获取事件循环并运行协程 loop = asyncio.get_event_loop() loop.run_until_complete(main())

2.2 协程(Coroutine)

协程是异步编程的基本单位，使用async def定义的函数就是协程。协程的特点是可以在执行过程中暂停，让出控制权给事件循环。

async def my_coroutine(): print('Start') await asyncio.sleep(1) # 模拟I/O操作 print('End')

2.3 await关键字

await用于挂起协程的执行，直到awaitable对象完成。它只能在协程内部使用。

async def fetch_data(): # 假设get_data是一个异步函数 data = await get_data() # 挂起当前协程，直到get_data完成 return data

3. 实战技巧：高效使用asyncio

3.1 并发执行多个任务

使用asyncio.gather()可以并发运行多个协程：

import asyncio async def fetch_url(url): print(f'Fetching {url}') await asyncio.sleep(2) # 模拟网络请求 print(f'Finished {url}') return f'Result from {url}' async def main(): urls = ['url1', 'url2', 'url3'] results = await asyncio.gather( *[fetch_url(url) for url in urls] ) print(results) asyncio.run(main())

3.2 超时控制

为异步操作设置超时时间：

async def slow_operation(): await asyncio.sleep(10) return 'Done' async def main(): try: result = await asyncio.wait_for(slow_operation(), timeout=5.0) except asyncio.TimeoutError: print('Operation timed out')

3.3 任务取消

可以取消正在运行的任务：

async def long_running_task(): try: while True: print('Working...') await asyncio.sleep(1) except asyncio.CancelledError: print('Task was cancelled') raise async def main(): task = asyncio.create_task(long_running_task()) await asyncio.sleep(3) task.cancel() try: await task except asyncio.CancelledError: print('Main caught cancellation') asyncio.run(main())

4. 常见陷阱与解决方案

4.1 阻塞代码破坏事件循环

在协程中调用同步阻塞代码会破坏事件循环：

# 错误示例 async def bad_example(): time.sleep(1) # 同步阻塞调用

解决方案是使用asyncio.to_thread()或loop.run_in_executor()：

async def good_example(): await asyncio.to_thread(time.sleep, 1) # 在单独线程中运行

4.2 忘记await

忘记await会导致协程不被执行：

# 错误示例 async def oops(): print('Start') asyncio.sleep(1) # 忘记await print('End') # 会立即执行

4.3 协程泄漏

创建任务但不保存引用可能导致协程泄漏：

# 错误示例 async def leaky(): for i in range(10): asyncio.create_task(some_task(i)) # 任务可能被GC回收

正确做法是保存任务引用：

async def proper(): tasks = [asyncio.create_task(some_task(i)) for i in range(10)] await asyncio.gather(*tasks)

5. 高级技巧与性能优化

5.1 使用异步上下文管理器

class AsyncResource: async def __aenter__(self): print('Acquiring resource') await asyncio.sleep(1) return self async def __aexit__(self, exc_type, exc, tb): print('Releasing resource') await asyncio.sleep(1) async def use_resource(): async with AsyncResource() as resource: print('Using resource') await asyncio.sleep(2) asyncio.run(use_resource())

5.2 限制并发数

使用信号量控制最大并发数：

async def worker(semaphore, task_id): async with semaphore: print(f'Task {task_id} started') await asyncio.sleep(2) print(f'Task {task_id} finished') async def main(): semaphore = asyncio.Semaphore(3) # 最多3个并发 tasks = [worker(semaphore, i) for i in range(10)] await asyncio.gather(*tasks) asyncio.run(main())

5.3 异步生成器

async def async_generator(): for i in range(5): await asyncio.sleep(1) yield i async def consume(): async for item in async_generator(): print(f'Got {item}') asyncio.run(consume())

6. 实际项目中的应用

6.1 异步Web爬虫

使用aiohttp实现高效爬虫：

import aiohttp import asyncio async def fetch_page(session, url): async with session.get(url) as response: return await response.text() async def crawl(urls): async with aiohttp.ClientSession() as session: tasks = [fetch_page(session, url) for url in urls] return await asyncio.gather(*tasks) # 示例使用 urls = ['https://example.com', 'https://example.org'] pages = asyncio.run(crawl(urls))

6.2 异步数据库访问

使用asyncpg连接PostgreSQL：

import asyncpg async def query_db(): conn = await asyncpg.connect('postgresql://user:pass@localhost/db') try: result = await conn.fetch('SELECT * FROM users WHERE id = $1', 1) print(result) finally: await conn.close() asyncio.run(query_db())

6.3 异步Web框架FastAPI

from fastapi import FastAPI import asyncio app = FastAPI() @app.get("/") async def read_root(): await asyncio.sleep(1) # 模拟I/O操作 return {"message": "Hello World"} @app.get("/items/{item_id}") async def read_item(item_id: int): await asyncio.sleep(0.5) return {"item_id": item_id}

7. 调试与测试异步代码

7.1 调试技巧

使用asyncio.debug模式：

async def buggy(): await asyncio.sleep(1) 1/0 # 故意制造错误 async def main(): try: await buggy() except Exception as e: print(f'Caught: {e}') # 启用调试模式 asyncio.run(main(), debug=True)

7.2 单元测试

使用pytest-asyncio插件：

import pytest @pytest.mark.asyncio async def test_async_code(): result = await some_async_function() assert result == expected_value

7.3 性能分析

使用cProfile分析异步代码：

import cProfile import asyncio async def task(): await asyncio.sleep(1) async def main(): await asyncio.gather(*[task() for _ in range(5)]) # 性能分析 cProfile.run('asyncio.run(main())', sort='cumtime')

8. 与其他技术的结合

8.1 异步与多进程结合

import concurrent.futures import asyncio def cpu_bound(number): return sum(i * i for i in range(number)) async def main(): with concurrent.futures.ProcessPoolExecutor() as pool: result = await asyncio.get_event_loop().run_in_executor( pool, cpu_bound, 10_000_000 ) print(result) asyncio.run(main())

8.2 异步与线程池结合

import asyncio import time def blocking_io(): time.sleep(1) return 'IO result' async def main(): loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, blocking_io) print(result) asyncio.run(main())

8.3 异步与同步代码的互操作

import asyncio import threading def sync_function(): print(f'Sync function in thread {threading.current_thread().name}') async def async_function(): print(f'Async function in thread {threading.current_thread().name}') await asyncio.sleep(1) async def main(): # 在协程中调用同步函数 sync_function() # 在同步代码中运行协程 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(async_function()) loop.close() asyncio.run(main())

9. 异步编程最佳实践

  1. 避免在协程中调用阻塞代码：使用asyncio.to_thread()或run_in_executor()包装阻塞调用

  2. 合理设置超时：为所有网络请求和外部调用设置超时

  3. 限制并发数：使用信号量或专门的限流工具控制并发请求数

  4. 正确处理异常：确保所有任务都有适当的异常处理

  5. 使用结构化并发：使用asyncio.TaskGroup(Python 3.11+)管理相关任务

  6. 监控和日志：为异步操作添加适当的日志记录

  7. 资源清理：确保所有资源(连接、文件等)在不再需要时被正确释放

  8. 性能测试：对异步代码进行压力测试，确保在高负载下表现良好

10. 常见问题解答

10.1 什么时候应该使用异步编程？

异步编程最适合I/O密集型应用，如：

  • Web服务器和客户端
  • 数据库访问
  • 网络爬虫
  • 微服务通信
  • 实时数据处理

对于CPU密集型任务，应考虑多进程或其他并行计算方案。

10.2 async/await和线程有什么区别？

  1. 协程是协作式多任务，线程是抢占式多任务
  2. 协程切换开销更小，因为不需要操作系统介入
  3. 协程避免了锁的需求，因为同一时间只有一个协程在执行
  4. 协程更容易调试，因为执行顺序更确定

10.3 如何选择异步库？

  1. 检查库是否原生支持asyncio
  2. 优先选择活跃维护的项目
  3. 查看社区评价和基准测试
  4. 确保API设计符合你的需求
  5. 考虑与其他工具的兼容性

一些推荐的异步库：

  • HTTP客户端: aiohttp, httpx
  • 数据库: asyncpg, databases, aioredis
  • Web框架: FastAPI, Sanic, Quart
  • 消息队列: aiokafka, aio-pika

10.4 如何处理异步代码中的共享状态？

  1. 尽量避免共享状态
  2. 使用asyncio.Lock保护共享资源
  3. 考虑使用actor模式
  4. 将共享状态封装在专门的管理类中
  5. 使用不可变数据结构

10.5 如何调试卡住的异步程序？

  1. 启用asyncio调试模式
  2. 使用asyncio.all_tasks()检查所有运行中的任务
  3. 添加超时和取消逻辑
  4. 使用日志记录关键步骤
  5. 逐步隔离问题代码

11. 实战项目：构建异步微服务

让我们构建一个简单的异步微服务，包含以下功能：

  • HTTP API端点
  • 数据库访问
  • 外部API调用
  • 后台任务

11.1 项目结构

async_microservice/ ├── main.py # 应用入口 ├── config.py # 配置 ├── database.py # 数据库连接 ├── models.py # 数据模型 ├── services.py # 业务逻辑 └── api.py # 路由和端点

11.2 数据库连接

# database.py import asyncpg from asyncpg.pool import Pool class Database: def __init__(self): self.pool: Pool = None async def connect(self, dsn: str): self.pool = await asyncpg.create_pool(dsn) async def disconnect(self): if self.pool: await self.pool.close() async def fetch_rows(self, query: str, *args): async with self.pool.acquire() as conn: return await conn.fetch(query, *args) async def execute(self, query: str, *args): async with self.pool.acquire() as conn: return await conn.execute(query, *args)

11.3 业务逻辑

# services.py import aiohttp from .models import User from .database import Database class UserService: def __init__(self, db: Database): self.db = db async def get_user(self, user_id: int) -> User: query = "SELECT * FROM users WHERE id = $1" row = await self.db.fetch_row(query, user_id) return User(**row) if row else None async def fetch_github_profile(self, username: str): async with aiohttp.ClientSession() as session: url = f"https://api.github.com/users/{username}" async with session.get(url) as resp: if resp.status == 200: return await resp.json() return None

11.4 API端点

# api.py from fastapi import FastAPI, HTTPException from .services import UserService from .database import Database app = FastAPI() db = Database() @app.on_event("startup") async def startup(): await db.connect("postgresql://user:pass@localhost/db") @app.on_event("shutdown") async def shutdown(): await db.disconnect() @app.get("/users/{user_id}") async def get_user(user_id: int): service = UserService(db) user = await service.get_user(user_id) if not user: raise HTTPException(status_code=404) return user @app.get("/github/{username}") async def get_github_profile(username: str): service = UserService(db) profile = await service.fetch_github_profile(username) if not profile: raise HTTPException(status_code=404) return profile

11.5 后台任务

# tasks.py import asyncio from .services import UserService from .database import Database async def background_task(db: Database, interval: int = 60): service = UserService(db) while True: print("Running background task...") # 这里可以执行定期任务，如清理过期数据等 await asyncio.sleep(interval) async def start_background_tasks(): db = Database() await db.connect("postgresql://user:pass@localhost/db") asyncio.create_task(background_task(db))

11.6 运行应用

# main.py import uvicorn from .api import app from .tasks import start_background_tasks if __name__ == "__main__": # 启动后台任务 asyncio.run(start_background_tasks()) # 启动FastAPI应用 uvicorn.run(app, host="0.0.0.0", port=8000)

12. 性能调优技巧

12.1 连接池优化

对于数据库和HTTP客户端，合理配置连接池大小：

# 数据库连接池配置 async def get_db_pool(): return await asyncpg.create_pool( dsn="postgresql://user:pass@localhost/db", min_size=5, # 最小连接数 max_size=20, # 最大连接数 max_queries=50000, # 单个连接最大查询次数 max_inactive_connection_lifetime=300 # 不活跃连接存活时间(秒) ) # HTTP客户端配置 async with aiohttp.ClientSession( connector=aiohttp.TCPConnector( limit=100, # 最大连接数 limit_per_host=10, # 单主机最大连接数 enable_cleanup_closed=True # 清理关闭的连接 ) ) as session: # 使用session

12.2 批量处理

将多个小操作合并为批量操作：

# 批量插入数据 async def batch_insert_users(db: Database, users: list): query = "INSERT INTO users (name, email) VALUES ($1, $2)" await db.executemany(query, [(u.name, u.email) for u in users])

12.3 内存优化

对于大数据处理，使用异步生成器避免内存爆炸：

async def stream_large_dataset(db: Database): async with db.pool.acquire() as conn: async with conn.transaction(): async for record in conn.cursor("SELECT * FROM large_table"): yield process_record(record)

12.4 CPU密集型任务优化

使用run_in_executor将CPU密集型任务卸载到线程池：

def cpu_intensive(data): # 执行CPU密集型计算 return result async def process_data(data): loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, cpu_intensive, data) return result

13. 异步编程的未来发展

Python的异步编程生态系统仍在快速发展中，以下是一些值得关注的趋势：

  1. 结构化并发：Python 3.11引入的asyncio.TaskGroup提供了更安全的并发管理方式

  2. 异步生成器改进：对异步生成器的性能优化和新特性支持

  3. 更好的调试工具：更强大的异步代码调试和分析工具

  4. 与其他语言的互操作：如通过PyO3与Rust的异步生态交互

  5. 更广泛的库支持：越来越多的库原生支持asyncio

  6. 性能优化：持续改进的事件循环实现和协程调度算法

  7. 标准库扩展：更多异步功能被加入Python标准库

  8. 教育资源的丰富：更多高质量的异步编程教程和最佳实践指南

14. 资源推荐

14.1 官方文档

  • asyncio官方文档
  • PEP 492 - Coroutines with async and await syntax
  • PEP 525 - Asynchronous Generators

14.2 书籍

  • "Python Concurrency with asyncio" by Matthew Fowler
  • "Using Asyncio in Python" by Caleb Hattingh
  • "Advanced Python Programming" by Dr. Gabriele Lanaro

14.3 视频教程

  • "Async Python from the Ground Up" by David Beazley
  • "Advanced asyncio: Solving Real-World Production Problems" by Lynn Root
  • "Asynchronous Python for Beginners" by Michael Kennedy

14.4 开源项目

  • FastAPI - 现代异步Web框架
  • aio-libs - 一系列高质量的异步库
  • uvicorn - 快速的ASGI服务器

15. 个人经验分享

在实际项目中使用异步编程多年，我总结了以下几点深刻体会：

  1. 渐进式采用：不要试图一次性将整个项目改为异步，可以从I/O密集的部分开始

  2. 监控是关键：异步应用的性能特征与同步应用不同，需要专门的监控

  3. 理解事件循环：深入理解事件循环的工作原理能帮助你写出更好的异步代码

  4. 避免过度并发：虽然异步可以轻松创建大量并发任务，但资源是有限的

  5. 测试挑战：异步代码的测试需要不同的方法，特别是涉及时间相关逻辑时

  6. 团队学习曲线：确保团队成员都理解异步编程的基本概念和陷阱

  7. 工具链成熟度：异步生态的工具链仍在发展中，某些场景可能需要自己造轮子

  8. 性能不是银弹：异步编程能提高I/O密集型应用的吞吐量，但不一定减少延迟

最令我印象深刻的一个教训是：在一次高负载场景下，我们没有限制对外部API的并发请求数，结果导致对方服务器过载，最终我们的服务也被限制访问。这个经历教会了我，异步编程赋予我们强大能力的同时，也要求我们更加负责任地使用这些能力。

编程学习 学习日记 实战经验
← 返回列表