Python aiohttp异步实现批量接口健康检查：限流并发与异常处理实战

脚本专家

在后端开发和运维工作中，批量检测接口存活、响应时间与状态码是日常刚需。传统的 requests 串行循环方案，100个接口即使每个超时3秒，总耗时也要300秒，无法满足高频监控需求。基于 asyncio + aiohttp 的异步方案可将 100 个接口的检查耗时压缩到单个接口的最大响应时间量级，效率提升数十倍。本文从零构建一套可配置、带异常捕获、支持并发限制的异步接口健康检查工具，可直接落地项目监控或定时巡检。

运行环境：Python 3.7+；核心依赖：aiohttp（安装命令：pip install aiohttp）。

一、异步为何适合健康检查？
接口健康检查是纯 IO 密集型场景，程序绝大多数时间在等待服务器响应，几乎不消耗 CPU。同步方案逐个请求必须等上一个完成才能执行下一个，规模越大耗时越长。异步方案利用事件循环在 IO 等待期间切换执行其他请求，实现多接口并发，同时可通过信号量（Semaphore）控制并发数，防止瞬间大量请求压垮目标服务器。

二、极简版：异步批量并发检查
下面的代码实现了最基础的异步健康检查：批量请求接口，记录状态、响应时间与异常信息。全局复用 aiohttp.ClientSession 减少 TCP 连接开销，全异常捕获保证单个接口报错不影响整体巡检。

2. import asyncio
3. import aiohttp
4. import time
6. API_LIST = [
7.     {"name": "百度首页", "url": "https://www.baidu.com", "method": "GET"},
8.     {"name": "HTTPBIN GET", "url": "https://httpbin.org/get", "method": "GET"},
9.     {"name": "HTTPBIN POST", "url": "https://httpbin.org/post", "method": "POST"},
10.     {"name": "无效接口", "url": "https://httpbin.org/error", "method": "GET"},
11. ]
13. TIMEOUT = aiohttp.ClientTimeout(total=5)
15. async def check_api(session: aiohttp.ClientSession, api: dict):
16.     start_time = time.time()
17.     name = api["name"]
18.     url = api["url"]
19.     method = api["method"]
20.     try:
21.         if method.upper() == "GET":
22.             async with session.get(url, timeout=TIMEOUT) as resp:
23.                 response_time = round((time.time() - start_time) * 1000, 2)
24.                 return {
25.                     "接口名称": name,
26.                     "接口地址": url,
27.                     "请求方式": method,
28.                     "状态": "正常",
29.                     "状态码": resp.status,
30.                     "响应耗时(ms)": response_time,
31.                     "异常信息": ""
32.                 }
33.         elif method.upper() == "POST":
34.             async with session.post(url, timeout=TIMEOUT) as resp:
35.                 response_time = round((time.time() - start_time) * 1000, 2)
36.                 return {
37.                     "接口名称": name,
38.                     "接口地址": url,
39.                     "请求方式": method,
40.                     "状态": "正常",
41.                     "状态码": resp.status,
42.                     "响应耗时(ms)": response_time,
43.                     "异常信息": ""
44.                 }
45.     except Exception as e:
46.         response_time = round((time.time() - start_time) * 1000, 2)
47.         return {
48.             "接口名称": name,
49.             "接口地址": url,
50.             "请求方式": method,
51.             "状态": "异常",
52.             "状态码": None,
53.             "响应耗时(ms)": response_time,
54.             "异常信息": str(e)
55.         }
57. async def main():
58.     async with aiohttp.ClientSession(timeout=TIMEOUT) as session:
59.         tasks = [check_api(session, api) for api in API_LIST]
60.         results = await asyncio.gather(*tasks)
61.         print("=" * 80)
62.         print("接口健康巡检结果")
63.         print("=" * 80)
64.         for res in results:
65.             print(f"接口：{res['接口名称']:10} | 状态：{res['状态']} | 状态码：{res['状态码']} | 耗时：{res['响应耗时(ms)']}ms | 异常：{res['异常信息']}")
67. if __name__ == "__main__":
68.     start = time.time()
69.     asyncio.run(main())
70.     total_time = round(time.time() - start, 2)
71.     print(f"本次巡检总耗时：{total_time}s")

复制代码

三、进阶优化：限流并发避免压垮服务
无限制并发在检测几百个接口时可能触发服务器限流或防火墙拦截。通过 asyncio.Semaphore 设置最大并发数，配合 aiohttp.TCPConnector 优化连接池，同时细化状态码校验规则。下面示例限制最多3个接口同时请求。

2. import asyncio
3. import aiohttp
4. import time
6. API_LIST = [
7.     {"name": "百度首页", "url": "https://www.baidu.com", "method": "GET"},
8.     {"name": "HTTPBIN GET", "url": "https://httpbin.org/get", "method": "GET"},
9.     {"name": "HTTPBIN POST", "url": "https://httpbin.org/post", "method": "POST"},
10.     {"name": "无效接口", "url": "https://httpbin.org/error", "method": "GET"},
11.     {"name": "测试接口1", "url": "https://httpbin.org/delay/1", "method": "GET"},
12.     {"name": "测试接口2", "url": "https://httpbin.org/delay/2", "method": "GET"},
13. ]
15. MAX_CONCURRENT = 3
16. TIMEOUT = aiohttp.ClientTimeout(total=5)
17. semaphore = asyncio.Semaphore(MAX_CONCURRENT)
19. async def check_api(session: aiohttp.ClientSession, api: dict):
20.     async with semaphore:
21.         start_time = time.time()
22.         name = api["name"]
23.         url = api["url"]
24.         method = api["method"]
25.         try:
26.             if method.upper() == "GET":
27.                 async with session.get(url, timeout=TIMEOUT) as resp:
28.                     response_time = round((time.time() - start_time) * 1000, 2)
29.                     status = "正常" if resp.status in [200, 201] else "异常"
30.                     return {
31.                         "接口名称": name,
32.                         "接口地址": url,
33.                         "状态": status,
34.                         "状态码": resp.status,
35.                         "响应耗时(ms)": response_time,
36.                         "异常信息": ""
37.                     }
38.             elif method.upper() == "POST":
39.                 async with session.post(url, timeout=TIMEOUT) as resp:
40.                     response_time = round((time.time() - start_time) * 1000, 2)
41.                     status = "正常" if resp.status in [200, 201] else "异常"
42.                     return {
43.                         "接口名称": name,
44.                         "接口地址": url,
45.                         "状态": status,
46.                         "状态码": resp.status,
47.                         "响应耗时(ms)": response_time,
48.                         "异常信息": ""
49.                     }
50.         except Exception as e:
51.             response_time = round((time.time() - start_time) * 1000, 2)
52.             return {
53.                 "接口名称": name,
54.                 "接口地址": url,
55.                 "状态": "异常",
56.                 "状态码": None,
57.                 "响应耗时(ms)": response_time,
58.                 "异常信息": str(e)
59.             }
61. async def main():
62.     connector = aiohttp.TCPConnector(limit=MAX_CONCURRENT)
63.     async with aiohttp.ClientSession(connector=connector, timeout=TIMEOUT) as session:
64.         tasks = [check_api(session, api) for api in API_LIST]
65.         results = await asyncio.gather(*tasks)
66.         normal_count = len([res for res in results if res["状态"] == "正常"])
67.         error_count = len(results) - normal_count
68.         print("=" * 90)
69.         print("接口健康巡检报告")
70.         print("=" * 90)
71.         for res in results:
72.             print(f"接口：{res['接口名称']:10} | 状态：{res['状态']} | 状态码：{res['状态码']} | 耗时：{res['响应耗时(ms)']}ms | 异常：{res['异常信息']}")
73.         print(f"正常接口：{normal_count} 个 | 异常接口：{error_count} 个")
75. if __name__ == "__main__":
76.     start = time.time()
77.     asyncio.run(main())
78.     print(f"本次巡检总耗时：{round(time.time() - start, 2)}s")

复制代码

四、企业级扩展：支持请求头、请求参数与POST传参
实际业务中多数接口需要 Token 鉴权、JSON 参数或表单提交。下面的配置示例支持自定义 headers 和 data（POST 时使用 json 参数发送），兼容 GET 与 POST 方法。

2. import asyncio
3. import aiohttp
4. import time
6. API_LIST = [
7.     {
8.         "name": "带鉴权GET接口",
9.         "url": "https://httpbin.org/headers",
10.         "method": "GET",
11.         "headers": {"Authorization": "Bearer test123456", "User-Agent": "Mozilla/5.0"},
12.         "data": None
13.     },
14.     {
15.         "name": "带参数POST接口",
16.         "url": "https://httpbin.org/post",
17.         "method": "POST",
18.         "headers": {"Content-Type": "application/json"},
19.         "data": {"username": "admin", "password": "123456"}
20.     }
21. ]
23. MAX_CONCURRENT = 3
24. TIMEOUT = aiohttp.ClientTimeout(total=5)
25. semaphore = asyncio.Semaphore(MAX_CONCURRENT)
27. async def check_api(session: aiohttp.ClientSession, api: dict):
28.     async with semaphore:
29.         start_time = time.time()
30.         name = api["name"]
31.         url = api["url"]
32.         method = api["method"]
33.         headers = api.get("headers", {})
34.         data = api.get("data")
35.         try:
36.             if method.upper() == "GET":
37.                 async with session.get(url, headers=headers, timeout=TIMEOUT) as resp:
38.                     response_time = round((time.time() - start_time) * 1000, 2)
39.                     return {
40.                         "接口名称": name, "状态": "正常" if resp.status == 200 else "异常",
41.                         "状态码": resp.status, "耗时(ms)": response_time, "异常": ""
42.                     }
43.             elif method.upper() == "POST":
44.                 async with session.post(url, headers=headers, json=data, timeout=TIMEOUT) as resp:
45.                     response_time = round((time.time() - start_time) * 1000, 2)
46.                     return {
47.                         "接口名称": name, "状态": "正常" if resp.status == 200 else "异常",
48.                         "状态码": resp.status, "耗时(ms)": response_time, "异常": ""
49.                     }
50.         except Exception as e:
51.             return {
52.                 "接口名称": name, "状态": "异常", "状态码": None,
53.                 "耗时(ms)": round((time.time() - start_time) * 1000, 2), "异常": str(e)
54.             }
56. async def main():
57.     connector = aiohttp.TCPConnector(limit=MAX_CONCURRENT)
58.     async with aiohttp.ClientSession(connector=connector, timeout=TIMEOUT) as session:
59.         tasks = [check_api(session, api) for api in API_LIST]
60.         results = await asyncio.gather(*tasks)
61.         for res in results:
62.             print(res)
64. if __name__ == "__main__":
65.     asyncio.run(main())

复制代码

五、常见问题避坑
1. 为什么不用 requests 做异步？
requests 是同步阻塞库，不兼容 asyncio 事件循环，强行在异步中使用会阻塞整个程序导致并发失效。必须使用 aiohttp。

2. 并发数设置多少合适？
内网接口可设置 10-20，公网接口建议 3-5，避免因 IP 封禁。根据目标服务器性能灵活调整。

3. 偶尔报连接超时怎么办？
属于正常网络波动，代码已捕获异常不会影响整体巡检。可适当增大 aiohttp.ClientTimeout 的 total 值。

六、定时巡检部署思路
可将上述 main 函数结合 APScheduler 实现秒级/分钟级定时执行，异常接口对接钉钉或企业微信机器人推送告警，实现无人值守监控。

总结：Python 异步 IO 实现的接口健康检查方案效率高、安全可控（信号量限流）、扩展性强（支持鉴权/参数/定时/告警），全局异常捕获保证了单个接口故障不影响整体。提供的代码可直接集成到个人项目或企业运维监控系统，快速搭建轻量化接口健康监控体系。

热心网友5

感谢分享！异步方案大幅提升批量接口检查效率，这个极简版代码结构很清晰，直接套用 `asyncio` + `aiohttp` 复用 `ClientSession` 的做法也很规范。不过实际生产里接口数量多时，如果不限制并发数可能会把目标服务器打满甚至触发限流，期待你在后续内容里加上 `Semaphore` 限流的实现。另外异常捕获可以再细化一下，比如区分 `asyncio.TimeoutError` 和 `aiohttp.ClientError`，这样排查异常原因会更方便。坐等限流并发与异常处理的完整实战篇！

热心网友5

楼主分享的这个异步健康检查方案太实用了！我之前一直用 requests 串行循环，几十个接口检查下来确实慢得崩溃。aiohttp 配合 asyncio 的思路非常清晰，全局复用 session 减少开销也很关键。不过标题里提到了“限流并发”，代码里好像只看到超时设置和异常捕获，没有信号量控制并发数的部分？如果能加上 Semaphore 限制同时发起的请求数，避免瞬时流量冲击目标服务器，应该会更完善。期待楼主后续把完整版贴出来，包括限流和结果汇总的代码，直接拿来当监控脚本用。

热心网友5

写得非常实用，正好最近在优化定时巡检脚本，之前用 requests 串行跑几十个接口确实慢得受不了。你这个异步方案加上超时和异常处理，直接就能应用到生产环境。 想请教两个细节：一是你用 `aiohttp.ClientTimeout(total=5)` 设置了整体超时，但有些接口可能需要区分连接超时和读取超时，有没有考虑过分别设置 `sock_connect` 和 `sock_read`？二是如果接口列表里混着 HTTP 和 HTTPS，`aiohttp` 是默认复用 Session 对连接池有啥影响吗？先谢过！