在Python开发中，当需要处理成百上千个网络请求时，同步阻塞的请求方式会导致程序大量时间浪费在等待网络响应上，整体效率极低。使用asyncio库可以基于事件循环实现异步并发，让单个线程就能同时管理多个网络连接的请求和响应，大幅提升资源利用率。
asyncio核心概念
要理解如何用asyncio优化并发连接管理，首先需要掌握几个核心概念：
- 事件循环：asyncio的核心调度器，负责监听和管理所有异步任务的执行，当某个任务等待IO时，事件循环会切换到其他可执行的任务。
- 协程：用async def定义的函数，是异步执行的基本单位，协程内部可以通过await挂起自身，让出执行权给事件循环。
- 任务：对协程的封装，事件循环可以直接调度任务执行，任务可以跟踪协程的执行状态。
基础异步请求实现
首先看一个使用asyncio配合aiohttp库实现异步网络请求的基础示例，aiohttp是支持异步操作的HTTP客户端库，和asyncio配合度很高。
import asyncio
import aiohttp
# 定义单个异步请求协程
async def fetch_url(session, url):
try:
# 发起异步GET请求
async with session.get(url) as response:
# 等待响应内容返回
content = await response.text()
print(f"请求{url}完成，响应长度：{len(content)}")
return content
except Exception as e:
print(f"请求{url}失败，错误：{e}")
return None
# 定义主协程，管理所有请求任务
async def main(url_list):
# 创建aiohttp客户端会话
async with aiohttp.ClientSession() as session:
# 创建所有请求任务
tasks = [asyncio.create_task(fetch_url(session, url)) for url in url_list]
# 等待所有任务完成
results = await asyncio.gather(*tasks)
return results
if __name__ == "__main__":
# 待请求的URL列表
urls = [
"http://ipipp.com/api/test1",
"http://ipipp.com/api/test2",
"http://ipipp.com/api/test3"
]
# 启动事件循环运行主协程
final_results = asyncio.run(main(urls))
优化并发连接管理
上面的基础示例没有限制并发数量，如果同时发起上千个请求，可能会给目标服务器造成过大压力，也可能因为本地资源不足导致请求失败。需要优化并发连接的管理策略。
限制最大并发数
可以使用asyncio的Semaphore（信号量）来限制同时执行的请求数量，避免并发过高带来的问题。
import asyncio
import aiohttp
# 定义最大并发数
MAX_CONCURRENCY = 10
# 创建信号量，控制并发数量
semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
async def fetch_url_with_limit(session, url):
# 获取信号量，超过并发数时会阻塞等待
async with semaphore:
try:
async with session.get(url) as response:
content = await response.text()
print(f"请求{url}完成，响应长度：{len(content)}")
return content
except Exception as e:
print(f"请求{url}失败，错误：{e}")
return None
async def main_with_limit(url_list):
async with aiohttp.ClientSession() as session:
tasks = [asyncio.create_task(fetch_url_with_limit(session, url)) for url in url_list]
results = await asyncio.gather(*tasks)
return results
if __name__ == "__main__":
urls = [f"http://ipipp.com/api/test{i}" for i in range(1, 51)]
asyncio.run(main_with_limit(urls))
配置连接池参数
aiohttp的ClientSession支持配置连接池参数，进一步精细化管理连接，比如限制总连接数、每个主机的连接数等。
import asyncio
import aiohttp
async def fetch_with_session_config(session, url):
try:
async with session.get(url) as response:
content = await response.text()
return content
except Exception as e:
print(f"请求错误：{e}")
return None
async def main_with_pool_config(url_list):
# 配置连接器参数，限制总连接数和每主机连接数
connector = aiohttp.TCPConnector(
limit=20, # 总连接数限制
limit_per_host=5, # 每个主机的最大连接数
ttl_dns_cache=300 # DNS缓存时间
)
# 创建带连接池配置的会话
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [asyncio.create_task(fetch_with_session_config(session, url)) for url in url_list]
results = await asyncio.gather(*tasks)
return results
if __name__ == "__main__":
urls = [f"http://ipipp.com/api/data{i}" for i in range(1, 101)]
asyncio.run(main_with_pool_config(urls))
异常处理与重试机制
处理大量网络请求时，网络波动、服务器超时等情况很常见，需要添加异常处理和重试机制提升请求成功率。
import asyncio
import aiohttp
# 最大重试次数
MAX_RETRY = 3
async def fetch_with_retry(session, url, retry_count=0):
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
# 如果响应状态码不是200，抛出异常触发重试
if response.status != 200:
raise Exception(f"状态码异常：{response.status}")
content = await response.text()
return content
except Exception as e:
# 如果未达到最大重试次数，递归重试
if retry_count < MAX_RETRY:
print(f"请求{url}失败，重试第{retry_count+1}次，错误：{e}")
await asyncio.sleep(1) # 重试前等待1秒
return await fetch_with_retry(session, url, retry_count+1)
else:
print(f"请求{url}达到最大重试次数，放弃请求")
return None
async def main_with_retry(url_list):
async with aiohttp.ClientSession() as session:
tasks = [asyncio.create_task(fetch_with_retry(session, url)) for url in url_list]
results = await asyncio.gather(*tasks)
return results
if __name__ == "__main__":
urls = [
"http://ipipp.com/api/stable",
"http://ipipp.com/api/unstable"
]
asyncio.run(main_with_retry(urls))
实践注意事项
- 不要在异步协程中使用同步阻塞的操作，比如time.sleep()，会阻塞整个事件循环，应该用asyncio.sleep()替代。
- 如果请求的是HTTPS地址，注意证书验证的问题，开发环境可以关闭验证，但生产环境建议开启。
- 对于需要登录或者携带固定请求头的场景，可以在创建ClientSession时统一配置headers参数，避免每个请求重复设置。
- 处理大量请求时，建议将结果分批处理，不要等所有请求都完成再统一处理，避免占用过多内存。