Python实现高延迟接口模拟方法

本文深入剖析了为何在Python异步测试中，仅用`asyncio.sleep`模拟高延迟接口存在严重缺陷——它仅挂起协程、无法复现真实网络调用中连接建立、TLS握手、服务端排队等不可控耗时，更无法体现延迟的非均匀分布特性（如P99高达2秒而多数请求极快），极易掩盖超时处理漏洞与并发竞争风险；想写出健壮可靠的异步代码？必须用更贴近生产环境的延迟建模方法。

asyncio.sleep 为什么不能直接替代真实网络延迟

因为 asyncio.sleep 只是挂起协程，不触发事件循环调度以外的 I/O；而真实接口延迟常伴随连接建立、TLS 握手、服务端排队等不可控耗时，这些 asyncio.sleep 完全模拟不了。测试中若只用它，可能掩盖超时逻辑缺陷或并发竞争问题。

• 真实高延迟接口往往有非均匀分布（如 P99 达 2s，但多数请求 asyncio.sleep 给固定值会漏掉毛刺场景
• 某些框架（如 aiohttp）在超时判断前会检查连接状态，单纯 sleep 不触发 socket 层行为，导致超时未生效
• 如果被测代码里有基于 asyncio.wait_for 的兜底逻辑，sleep 时间略短于 timeout 值时，测试会“侥幸通过”，但线上仍可能失败

如何用 asyncio.sleep 模拟更贴近真实的延迟分布

别写死一个数字，用随机 + 分位数组合逼近真实响应时间分布。例如按日志统计出该接口的 p50=120ms、p90=850ms、p99=2100ms，再用 random.uniform 或 random.choices 抽样。

• 简单分层模拟：random.choices([0.1, 0.8, 2.1], weights=[50, 40, 10])[0] —— 表示 50% 请求 ≤100ms，40% 在 100–850ms，10% ≥2s
• 避免用 time.sleep：它会阻塞整个事件循环，所有并发任务都停摆，和异步初衷相悖
• 注意单位：asyncio.sleep 接收秒为单位的 float，传 100 是 100 秒，不是毫秒；常用写法是 asyncio.sleep(0.12)

测试中 sleep 放错位置会导致并发行为失真

常见错误是把 asyncio.sleep 写在协程最开头，让所有并发请求“齐步走”——这跟真实接口的请求到达时间和处理并行性完全不符。

• 正确做法：在发起请求前、或 mock 返回前插入 sleep，比如 await asyncio.sleep(delay); return mock_response
• 如果测试目标是压测连接池耗尽，sleep 应放在获取连接之后、发送请求之前，否则连接池不会被占满
• 用 asyncio.create_task 启动多个协程时，确保每个 task 自己控制 delay，而不是统一 await 一次 sleep 再批量 create_task

mock 异步函数时怎么安全注入 sleep

直接 patch 一个返回 await asyncio.sleep 的协程容易出错，因为 patch 目标通常是同步函数，或没处理好协程返回类型。

• 推荐用 AsyncMock（Python 3.8+）：mock_func = AsyncMock(side_effect=lambda: asyncio.sleep(0.5) or {"data": "ok"})
• 若用 unittest.mock.patch，必须设 new_callable=AsyncMock，否则调用会报 TypeError: object MagicMock can't be used in 'await' expression
• 不要在 side_effect 里写 await asyncio.sleep(...)，因为 side_effect 执行在同步上下文；应改用 return 一个协程对象，或用 AsyncMock(return_value=...) 配合外部 sleep

真实接口的延迟不只是“慢”，还带抖动、失败率、连接中断等维度；sleep 只能补时间维度，其他得靠更重的 mock 工具（如 respx 或 httpx.MockTransport）配合使用。

今天带大家了解了的相关知识，希望对你有所帮助；关于文章的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~