aiohttp异步编程实战：并发请求与异步数据库连接池优化

脚本专家

在 Python 网络应用里，异步编程往往是性能优化中收益最直接的一环。原文作者提到过一个日均请求量超过百万的 API 网关项目：从同步 Flask 架构迁移到异步 aiohttp 之后，服务器资源消耗下降了 70%，平均响应时间从 200ms 缩短到 50ms。这个对比并非玄学，根源在于同步模型把时间都花在了等待 I/O 上，而异步模型把这些等待时间让给了其他请求。下面按事件循环、HTTP 客户端与服务端、异步数据库、WebSocket 的顺序，把原文的代码实现和参数含义重新梳理一遍。

一、同步与异步的差距是怎么量出来的

原文用一个最小可运行脚本做了对比：对同一批 5 个延迟 1 秒的接口，同步版用 requests 顺序请求，异步版用 aiohttp 的 ClientSession 并发请求，最后打印两者的耗时和提升倍数。这段脚本的价值在于它把“阻塞”和“非阻塞”变成了可观测的数字。

2. import time
3. import asyncio
4. import aiohttp
5. import requests
7. def sync_http_requests():
8.     urls = ['https://httpbin.org/delay/1'] * 5
9.     start = time.time()
10.     for url in urls:
11.         response = requests.get(url)
12.         print(f"同步请求完成: {response.status_code}")
13.     sync_time = time.time() - start
14.     print(f"同步总耗时: {sync_time:.2f}秒")
15.     return sync_time
17. async def async_http_requests():
18.     urls = ['https://httpbin.org/delay/1'] * 5
19.     start = time.time()
20.     async with aiohttp.ClientSession() as session:
21.         tasks = [session.get(url) for url in urls]
22.         responses = await asyncio.gather(*tasks)
23.         for i, response in enumerate(responses):
24.             print(f"异步请求 {i} 完成: {response.status}")
25.             await response.release()
26.     async_time = time.time() - start
27.     print(f"异步总耗时: {async_time:.2f}秒")
28.     return async_time
30. async def performance_comparison():
31.     sync_time = sync_http_requests()
32.     async_time = await async_http_requests()
33.     print(f"同步耗时: {sync_time:.2f}秒")
34.     print(f"异步耗时: {async_time:.2f}秒")
35.     print(f"性能提升: {sync_time/async_time:.1f}倍")
37. if __name__ == "__main__":
38.     asyncio.run(performance_comparison())

复制代码

两个细节值得注意。其一，aiohttp 的 session.get() 返回的是可等待的请求上下文，放进列表交给 asyncio.gather 才会真正并发发起；其二，拿到响应后要调用 await response.release()，把连接及时归还连接池，否则连接数会被慢慢占满。

二、asyncio 事件循环：调度器的三个关键动作

事件循环是异步程序的发动机，它负责决定哪个协程先跑、哪个 I/O 完成后唤醒谁。原文用 get_running_loop() 拿到当前循环，并演示了 call_soon（尽快执行同步回调）、call_later（延迟执行回调）以及 create_task + gather 的任务管理方式。理解这三点，基本就能看懂异步框架的行为。

2. import asyncio
4. class EventLoopInsight:
5.     @staticmethod
6.     async def demonstrate_event_loop():
7.         loop = asyncio.get_running_loop()
8.         print(f"事件循环: {loop}")
9.         print(f"循环时间: {loop.time()}")
10.         print(f"是否运行: {loop.is_running()}")
12.         def synchronous_callback():
13.             print("同步回调执行")
14.         loop.call_soon(synchronous_callback)
16.         def delayed_callback():
17.             print("延迟回调执行")
18.         loop.call_later(2, delayed_callback)
20.         await asyncio.sleep(1)
21.         print("主协程继续执行")
23.     @staticmethod
24.     async def task_management_demo():
25.         async def worker(name, duration):
26.             print(f"任务 {name} 开始执行")
27.             await asyncio.sleep(duration)
28.             print(f"任务 {name} 完成")
29.             return f"{name}-结果"
31.         tasks = [
32.             asyncio.create_task(worker("任务A", 2)),
33.             asyncio.create_task(worker("任务B", 1)),
34.             asyncio.create_task(worker("任务C", 3))
35.         ]
36.         results = await asyncio.gather(*tasks)
37.         print(f"所有任务完成: {results}")

复制代码

这里有个容易踩的点：call_later 注册的延迟回调不会阻塞主协程，主协程仍然按 await asyncio.sleep(1) 的节奏继续往下走；create_task 只是把协程包装成任务排进循环，真正等待结果的是 gather。

三、aiohttp 客户端：连接池与超时是两个必须显式配置的参数

默认配置在低并发下够用，到了高并发就必须调连接池和超时。原文给出的写法是 TCPConnector 控制连接数量，ClientTimeout 控制各类超时，两层配置分工明确：limit 是全局最大连接数，limit_per_host 限制单主机连接数，keepalive_timeout 决定空闲连接保留多久；timeout 一侧 total 是总超时，connect 是建连超时，sock_read 是读取超时。

2. import aiohttp
4. connector = aiohttp.TCPConnector(
5.     limit=100,
6.     limit_per_host=30,
7.     keepalive_timeout=30
8. )
9. timeout = aiohttp.ClientTimeout(
10.     total=60,
11.     connect=10,
12.     sock_read=30
13. )
15. async def client_demo():
16.     async with aiohttp.ClientSession(
17.         connector=connector,
18.         timeout=timeout,
19.         headers={'User-Agent': 'MyAsyncApp/1.0'}
20.     ) as session:
21.         urls = [
22.             'https://httpbin.org/json',
23.             'https://httpbin.org/uuid',
24.             'https://httpbin.org/headers'
25.         ]
26.         tasks = [session.get(url) for url in urls]
27.         responses = await asyncio.gather(*tasks)
28.         for i, response in enumerate(responses):
29.             data = await response.json()
30.             print(f"响应 {i}: {len(str(data))} 字节")

复制代码

如果只设 total 不设 sock_read，遇到对端迟迟不返回数据的情况，请求会一直挂到总超时才失败，排查时表现为“连接是通的，但请求永远不结束”。原文把三个超时拆开配置，正是为了避免这种互相牵连。

四、aiohttp 服务端：路由 + WebSocket 处理器

服务端部分由 web.Application 组织，通过 app.router.add_get 把路径绑定到协程处理器；普通接口用 web.Response 或 web.json_response 返回，WebSocket 接口用 web.WebSocketResponse 完成握手后异步迭代消息。

2. import time
3. import aiohttp
4. from aiohttp import web
6. async def handle_root(request):
7.     return web.Response(text="Hello, Async World!")
9. async def handle_api(request):
10.     data = {'status': 'ok', 'timestamp': time.time()}
11.     return web.json_response(data)
13. async def handle_websocket(request):
14.     ws = web.WebSocketResponse()
15.     await ws.prepare(request)
16.     async for msg in ws:
17.         if msg.type == aiohttp.WSMsgType.TEXT:
18.             await ws.send_str(f"ECHO: {msg.data}")
19.         elif msg.type == aiohttp.WSMsgType.ERROR:
20.             print(f"WebSocket错误: {ws.exception()}")
21.     return ws
23. app = web.Application()
24. app.router.add_get('/', handle_root)
25. app.router.add_get('/api', handle_api)
26. app.router.add_get('/ws', handle_websocket)
27. # web.run_app(app, host='127.0.0.1', port=8080)

复制代码

WSMsgType.ERROR 分支里取 ws.exception() 打印，是排查 WebSocket 断连原因最省事的办法：客户端异常断开、协议错误都会从这里冒出来，而不是静默断开。

五、异步数据库：连接池、事务和批量写入

databases 库的 Database 对象同时承担连接池和查询入口两个角色，min_size / max_size 就是池的上下水位。原文的 AsyncDatabaseManager 封装了 connect、建表、复杂查询与基准测试，其中最能体现异步收益的是“顺序插入”和“事务内批量插入”的对比。

2. from databases import Database
4. class AsyncDatabaseManager:
5.     def __init__(self, database_url, min_connections=2, max_connections=20):
6.         self.database_url = database_url
7.         self.min_connections = min_connections
8.         self.max_connections = max_connections
9.         self.db = None
11.     async def connect(self):
12.         self.db = Database(
13.             self.database_url,
14.             min_size=self.min_connections,
15.             max_size=self.max_connections
16.         )
17.         await self.db.connect()
18.         await self._create_tables()
20.     async def _create_tables(self):
21.         create_table_query = """
22.         CREATE TABLE IF NOT EXISTS users (
23.             id SERIAL PRIMARY KEY,
24.             username VARCHAR(50) UNIQUE NOT NULL,
25.             email VARCHAR(100) UNIQUE NOT NULL,
26.             created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
27.             updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
28.         );
29.         CREATE TABLE IF NOT EXISTS messages (
30.             id SERIAL PRIMARY KEY,
31.             user_id INTEGER REFERENCES users(id),
32.             content TEXT NOT NULL,
33.             created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
34.         );
35.         """
36.         await self.db.execute(create_table_query)
38.     async def perform_benchmark(self, iterations=1000):
39.         async def sequential_insert():
40.             start_time = time.time()
41.             for i in range(iterations):
42.                 query = "INSERT INTO users (username, email) VALUES (:username, :email)"
43.                 values = {"username": f"user_{i}", "email": f"user_{i}@example.com"}
44.                 await self.db.execute(query, values)
45.             return time.time() - start_time
47.         async def batch_insert():
48.             start_time = time.time()
49.             values_list = [
50.                 {"username": f"batch_user_{i}", "email": f"batch_{i}@example.com"}
51.                 for i in range(iterations)
52.             ]
53.             async with self.db.transaction():
54.                 query = "INSERT INTO users (username, email) VALUES (:username, :email)"
55.                 for values in values_list:
56.                     await self.db.execute(query, values)
57.             return time.time() - start_time
59.         sequential_time = await sequential_insert()
60.         batch_time = await batch_insert()
61.         print(f"顺序插入耗时: {sequential_time:.2f}秒")
62.         print(f"批量插入耗时: {batch_time:.2f}秒")
63.         print(f"性能提升: {sequential_time/batch_time:.1f}倍")
64.         await self.db.execute("DELETE FROM users WHERE username LIKE 'user_%' OR username LIKE 'batch_user_%'")
66.     async def disconnect(self):
67.         if self.db:
68.             await self.db.disconnect()

复制代码

事务块 async with self.db.transaction() 的作用不只是提速：把一批写入包在同一个事务里，可以减少提交次数，同时保证要么全成功、要么全回滚。原文的 complex_query_example 用事务先插用户、再插 5 条消息，最后用 JOIN 查询验证，就是典型的“写一致 + 读关联”组合用法。

六、SQLAlchemy 异步引擎：读写分离与连接池监控

当项目已经使用 ORM 时，可以换成 create_async_engine 走异步驱动。原文的 MultiDatabaseManager 同时维护主库和从库两个引擎，用 read_only 参数决定从哪个引擎取会话，并用 @asynccontextmanager 统一处理 commit 和 rollback。

2. from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
3. from contextlib import asynccontextmanager
5. class MultiDatabaseManager:
6.     def __init__(self, primary_db_url, replica_db_url=None):
7.         self.primary_engine = create_async_engine(
8.             primary_db_url,
9.             echo=False,
10.             pool_size=10,
11.             max_overflow=20,
12.             pool_pre_ping=True
13.         )
14.         self.replica_engine = None
15.         if replica_db_url:
16.             self.replica_engine = create_async_engine(
17.                 replica_db_url,
18.                 echo=False,
19.                 pool_size=5,
20.                 max_overflow=10,
21.                 pool_pre_ping=True
22.             )
24.     @asynccontextmanager
25.     async def get_session(self, read_only=False):
26.         engine = self.replica_engine if read_only and self.replica_engine else self.primary_engine
27.         async with AsyncSession(engine) as session:
28.             try:
29.                 yield session
30.                 await session.commit()
31.             except Exception:
32.                 await session.rollback()
33.                 raise
35.     async def connection_pool_metrics(self):
36.         primary_pool = self.primary_engine.pool
37.         print(f"连接池大小: {primary_pool.size()}")
38.         print(f"已检入连接: {primary_pool.checkedin()}")
39.         print(f"已检出连接: {primary_pool.checkedout()}")
40.         print(f"溢出连接: {primary_pool.overflow()}")

复制代码

这里的参数含义需要明确：pool_size 是常驻连接数，max_overflow 是高峰时允许临时超出的连接数，pool_pre_ping=True 会在取连接时先探活，避免拿到已失效的旧连接报错。connection_pool_metrics 打印的 checkedout 与 overflow 是判断“是不是连接池不够用”的直接依据——如果 checkedout 长期贴着上限、overflow 持续增长，就该考虑调大 pool_size 或者缩短会话持有时间。

七、WebSocket 连接管理

实时推送场景下，服务端需要维护“谁在线、往谁发”的映射关系。原文用 WebSocketManager 类持有 connections 字典（键为标识，值为 WebSocketResponse 集合），这也是异步框架里管理长连接的常见结构：连接建立时登记，断开时移除，广播时遍历集合逐个 send_str。处理消息循环时用 async for msg in ws 接收，并按 WSMsgType.TEXT / ERROR 分支处理，与上面服务端示例的写法一致。

八、优化与排查清单

按原文涉及的技术点归纳，落地时可以按这几条对照：

1. 客户端连接池要设上限。limit 与 limit_per_host 不设，高并发下容易把连接数打满或触发对端限流。
2. 三类超时分开配。total、connect、sock_read 各管一段，排查“卡住不返回”时先看 sock_read。
3. 响应记得 release。并发请求拿到结果后释放连接，让连接回到池里复用。
4. 批量写入包事务。事务既能减少提交开销，也能在异常时统一回滚。
5. 连接池指标要能看见。size / checkedin / checkedout / overflow 四个数字足以定位池是否耗尽。
6. 长连接异常要打印 exception。WebSocket 的 ERROR 分支不取 ws.exception()，线上断连基本无从下手。
7. 生产环境关掉 SQL echo。原文在 create_async_engine 里显式写了 echo=False 并注明生产环境应如此，避免日志把 SQL 全量刷出来拖慢性能。

整体来看，原文的代码骨架覆盖了异步网络应用的主要环节：事件循环负责调度，aiohttp 负责收发，异步数据库驱动负责持久化，WebSocket 负责实时推送。把这些组件的参数配好、把连接与事务管好，同步架构迁移到异步架构时所说的“十倍级提升”才有现实基础；反过来，如果连接池不设上限、超时不分开、连接不释放，即使换成了 asyncio，也只是把阻塞换了个地方发生。

热心网友5

感谢分享，这篇把同步和异步的差距用 5 个延迟接口量化出来，很直观。两个细节抓得很准：session.get 放进 gather 才真正并发，response.release 及时归还连接池，这两点新手很容易漏。事件循环部分用 call_soon、call_later、create_task 加 gather 讲调度，思路清楚，适合作为实战前的概念铺垫。有个小提醒：对比脚本里同步 requests 直接放在异步流程里跑，用来测耗时没问题，但如果在异步服务里混用同步阻塞调用，会把事件循环卡住，实际项目里最好放线程池或独立进程。另外标题里的异步数据库连接池优化还没展开，看到任务管理那段好像也截断了，期待后续把连接池参数、超时和异常处理补上。整体很有参考价值。

热心网友5

感谢分享，这个对比脚本很适合做入门演示，同步/异步的差距一眼就能看到。想补充几个实战里容易踩的点，供参考： 一是公平性。同步版在循环里每次 requests.get 都是新建连接，TCP 和 TLS 握手的开销会算进同步耗时里；而异步版一个 ClientSession 天然复用连接池。如果想让对比更纯粹反映“阻塞等待”的差别，同步版可以改用 requests.Session() 复用连接，这样 5 个 1 秒延迟的接口同步约 5 秒、异步约 1 秒多，倍数关系会更接近理论值。 二是 response.release() 的位置。当前写法基本能工作，但更清晰的是 async with session.get(url) as resp 读取完再用，避免拿到状态码后忘了读 body 或异常分支漏释放。另外 gather 里如果某个请求抛异常，默认会往外冒，其余任务不会自动取消，实际项目里建议用 return_exceptions=True 或者对每个任务包一层 try。 三是并发不要无上限。帖子里 5 个 URL 没问题，但真实场景下如果几百上千个请求直接 gather，会瞬间打满连接池甚至被对端限流。常见做法是 asyncio.Semaphore 限制并发数，或者 TCPConnector(limit=..., limit_per_host=...) 配合，效果比单纯加任务更可控。 事

热心网友5

看完这篇梳理，感觉把异步的性能账算得很清楚，那个 200ms 到 50ms 的对比很有说服力。几个点我比较有共鸣。 第一，最开始的对比脚本确实是最直观的入门材料。session.get 返回的可等待对象、交给 gather 才真正并发，这个机制很多人第一次会踩坑，写成逐个 await 就退化成同步了。另外我平时更习惯用 async with session.get 作为上下文管理器，比手动 await response.release() 少一次忘记归还连接的机会，不过楼主点出 release 的意义是对的，连接池被占满的 bug 往往很隐蔽。 第二，建议做对比测试时把 httpbin 换成自己本地起的延迟接口，外网波动有时会让“提升倍数”看起来忽高忽低，本地稳定环境测出来的数字更有参考价值。还有 ClientSession 尽量全局复用一个，每个请求新建 session 会把连接池优势抵消掉，楼主这个例子里只建一次是对的，后面如果扩成多个函数要注意传参。 第三，事件循环那段讲 call_soon、call_later、create_task 加 gather 三个动作，确实是读异步框架源码的钥匙。小提醒是 call_later 的回调返回的 handle 可以 cancel，做超时清理时用得上。 帖子后半截好像断在任务管理那里了，异步数据库连接池优化和 WebSocket 部分还