当前位置：首页 > news >正文

Python异步编程实战：利用asyncio处理高并发请求

news 2026/5/11 3:41:59

在当今互联网应用中，高并发请求处理已成为后端开发的核心挑战之一。Python的asyncio库为开发者提供了一套完整的异步I/O解决方案，能够显著提升程序的并发处理能力。本文将深入探讨asyncio的核心概念，并通过实战代码演示如何利用它处理高并发场景，同时穿插相关面试题解析。

异步编程基础概念

同步 vs 异步

同步编程中，代码按顺序执行，每个操作必须等待前一个操作完成后才能开始。而异步编程允许程序在等待I/O操作（如网络请求、文件读写）时继续执行其他任务，从而充分利用CPU资源。

事件循环（Event Loop）

事件循环是asyncio的核心，它负责调度和执行异步任务。当遇到await表达式时，事件循环会暂停当前协程，转而执行其他可运行的协程。

协程（Coroutine）

协程是asyncio中的基本执行单元，使用async def定义。协程可以通过await关键字挂起自身，让出控制权给事件循环。

asyncio核心组件实战

创建和运行协程

import asyncioasync def fetch_data(url):print(f"开始获取 {url}")await asyncio.sleep(2)  # 模拟网络请求延迟print(f"完成获取 {url}")return f"{url}的数据"async def main():# 创建任务列表tasks = [fetch_data("https://api.example.com/data1"),fetch_data("https://api.example.com/data2"),fetch_data("https://api.example.com/data3")]# 并发执行所有任务results = await asyncio.gather(*tasks)print(f"所有结果: {results}")# 运行主协程
asyncio.run(main())

使用Semaphore控制并发数

在高并发场景中，无限制地创建并发任务可能导致资源耗尽。Semaphore可以帮助我们控制同时运行的任务数量。

import asyncioclass RateLimiter:def __init__(self, max_concurrent):self.semaphore = asyncio.Semaphore(max_concurrent)async def fetch_with_limit(self, url):async with self.semaphore:print(f"开始处理 {url}")await asyncio.sleep(1)  # 模拟请求print(f"完成处理 {url}")return f"{url}的响应"async def main():limiter = RateLimiter(3)  # 最多同时3个请求urls = [f"https://api.example.com/item{i}" for i in range(10)]tasks = [limiter.fetch_with_limit(url) for url in urls]results = await asyncio.gather(*tasks)print(f"处理了 {len(results)} 个请求")asyncio.run(main())

面试题解析

面试题1：asyncio.run() 和 loop.run_until_complete() 的区别

问题：请解释asyncio.run()和loop.run_until_complete()的主要区别及使用场景。

解析：

asyncio.run()是Python 3.7引入的高级API，它创建新的事件循环，运行指定的协程，然后关闭事件循环。它简化了异步程序的入口点，适合大多数简单场景。

loop.run_until_complete()需要手动创建和管理事件循环，提供更细粒度的控制，适合需要复用事件循环或进行复杂配置的场景。

面试题2：async with 和 async for 的作用

问题：请说明async with和async for在异步编程中的作用，并给出示例。

解析：

async with用于异步上下文管理器，确保资源在异步操作中正确获取和释放。例如，数据库连接池的管理：

import asyncpg
import asyncioasync def query_database():# 在实际开发中，可以使用dblens SQL编辑器来设计和测试复杂的SQL查询# dblens提供了直观的界面和强大的调试功能，大大提升开发效率conn = await asyncpg.connect('postgresql://user:password@localhost/db')async with conn.transaction():result = await conn.fetch('SELECT * FROM users WHERE active = $1', True)await conn.close()return result

async for用于异步迭代器，逐项处理异步生成的数据流：

async def async_generator():for i in range(5):await asyncio.sleep(0.5)yield iasync def main():async for item in async_generator():print(f"收到: {item}")

实战：高并发HTTP请求处理

下面是一个完整的示例，展示如何使用aiohttp库处理高并发HTTP请求：

import aiohttp
import asyncio
import timeasync def fetch_url(session, url, semaphore):async with semaphore:try:async with session.get(url, timeout=10) as response:data = await response.text()return {'url': url,'status': response.status,'size': len(data)}except Exception as e:return {'url': url, 'error': str(e)}async def main():# 控制并发数为10semaphore = asyncio.Semaphore(10)# 准备URL列表urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 101)]start_time = time.time()async with aiohttp.ClientSession() as session:tasks = [fetch_url(session, url, semaphore) for url in urls]results = await asyncio.gather(*tasks)elapsed = time.time() - start_time# 统计结果success = sum(1 for r in results if 'status' in r and r['status'] == 200)print(f"总请求数: {len(urls)}")print(f"成功请求: {success}")print(f"总耗时: {elapsed:.2f}秒")print(f"平均每秒处理: {len(urls)/elapsed:.2f}个请求")# 在处理大量数据时，可以使用QueryNote记录和分析结果# QueryNote是dblens旗下的笔记工具，特别适合保存和分享技术分析结果return resultsif __name__ == "__main__":asyncio.run(main())

性能优化技巧

1. 连接池复用

对于频繁的网络请求，使用连接池可以避免重复建立连接的开销。aiohttp的ClientSession内部已经实现了连接池。

2. 适当调整并发限制

根据目标服务器的承受能力和网络带宽，合理设置Semaphore的值。过高的并发可能导致服务器拒绝服务或自身资源耗尽。

3. 超时设置

为每个请求设置合理的超时时间，避免因个别慢请求阻塞整个程序。

4. 错误处理与重试

from tenacity import retry, stop_after_attempt, wait_exponential
import aiohttp@retry(stop=stop_after_attempt(3),wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def fetch_with_retry(session, url):async with session.get(url) as response:return await response.text()