简介DaoyiHttp 是一款面向开发与测试人员的 HTTP 服务端与客户端双向模拟测试工具基于 C# 实现适合需要调试接口、验证协议行为或进行自动化测试的中初级开发者。它既能以客户端身份发起 GET、POST、PUT、DELETE 等请求支持自定义请求头、请求体、超时与重试策略也能作为服务端预设状态码、响应头与响应体模拟不同返回场景便于检验客户端兼容性与错误处理能力。资源包共 48 个文件以 16 个 png 与 7 个 jpg 界面截图、12 个 cs 源码文件为主另含 resx 资源、sln 解决方案、csproj 工程文件及 dll、pdb 等编译产物整体约 8.65MB目录按 UI、Http、Doc、Util 等模块划分结构清晰。目前已有 314 人学习下载。借助完整源码与工程配置读者可快速理解 HTTP 通信的请求响应流程掌握服务端模拟与客户端测试的实现思路并在此基础上二次开发或用于接口联调与压力测试场景。1. 一个工具同时扮演 HTTP 服务端和客户端为什么值得自己搭一套调接口时最烦的场景不是报错而是你根本不知道错在哪一端。前端说后端返回了 502后端说本地 curl 是通的运维说网关日志没记录——三方各执一词最后靠猜。一个能同时扮演 HTTP 服务端和客户端的测试工具价值就在这里它把「请求怎么发出去」和「请求怎么被接住」放进同一个进程里链路两端都握在自己手上中间没有黑匣子。这类工具解决的核心问题有三个。第一是自测闭环写完一个接口不用等联调自己起服务端、自己发客户端请求几秒内看到完整往返。第二是协议验证HTTP 连接复用、chunked 传输、超时重试这些行为只有自己控制两端才能精确复现。第三是故障注入服务端可以故意返回 502、延迟 3 秒、提前断连客户端可以故意发畸形头、超大 body用来验证对端的健壮性。适合谁用后端开发自测接口、测试工程师构造边界用例、运维排查网关行为甚至嵌入式方向做 STM32 HTTP 库对接时也需要一个可控的对端来验证。下面按「先立住原理、再动手复现、最后讲坑」的顺序拆开讲。2. 服务端与客户端同进程先想清楚架构再写第一行代码2.1 为什么不用现成的 Postman 或 curl现成工具能发请求但发出去之后对端是别人的服务你控制不了它怎么回。curl 加-v能看到请求头但看不到服务端内部的解析过程Postman 能 mock但 mock 规则和真实服务端行为有差距。自己写一个工具本质是把「可控性」拿回来服务端想返回什么状态码就返回什么客户端想发什么畸形报文就发什么。另一个现实原因是 HTTP 连接复用。很多线上问题出在 keep-alive 连接被中间层提前关闭用现成工具很难稳定复现因为连接池行为不透明。自己写的客户端可以精确控制连接建立、复用、关闭的时机配合自己写的服务端记录每条连接的生命周期问题就变成可观测的。2.2 同进程双角色的最小架构核心思路是一个进程内起两个线程或协程一个跑 HTTP server 监听端口一个跑 HTTP client 发请求。两者通过内存里的日志队列共享观测数据这样一次往返的请求头、响应头、耗时、连接 ID 都能对齐。用 Python 标准库就能搭出最小版本不引入额外依赖# dual_http_tool.py import http.server import threading import http.client import time import json # 共享日志记录服务端收到的每条请求 server_log [] class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): # 记录请求元信息供客户端侧对齐 entry { path: self.path, headers: dict(self.headers), client: self.client_address, ts: time.time(), } server_log.append(entry) body json.dumps({ok: True, path: self.path}).encode() self.send_response(200) self.send_header(Content-Type, application/json) self.send_header(Content-Length, str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, *args): pass # 关掉默认 stderr 输出避免干扰 def start_server(port): srv http.server.HTTPServer((127.0.0.1, port), Handler) t threading.Thread(targetsrv.serve_forever, daemonTrue) t.start() return srv def client_get(port, path): conn http.client.HTTPConnection(127.0.0.1, port, timeout5) t0 time.time() conn.request(GET, path) resp conn.getresponse() data resp.read() cost (time.time() - t0) * 1000 conn.close() return resp.status, data, cost if __name__ __main__: srv start_server(18080) status, data, cost client_get(18080, /health) print(fstatus{status} cost{cost:.1f}ms body{data.decode()}) print(server saw:, server_log) srv.shutdown()这段代码的逻辑说明Handler继承BaseHTTPRequestHandler重写do_GET处理 GET 请求把请求头、来源地址、时间戳存进server_log。start_server在守护线程里跑serve_forever主线程继续执行客户端逻辑。client_get用http.client建连接、发请求、读响应记录耗时。参数说明port选 18080 是为了避开常见的 8080、8000 冲突timeout5是客户端读超时防止服务端不响应时卡死daemonTrue保证主线程退出时服务端线程跟着结束不用手动清理。log_message被重写为空是因为默认实现会往 stderr 打日志在批量测试时刷屏。2.3 把两端观测数据对齐才有意义上面代码跑完会打印两行客户端看到的状态码和耗时服务端看到的请求记录。真正有用的是把这两者按时间戳对齐。比如客户端报告耗时 3000ms服务端记录显示请求在 10ms 内就处理完了那 2990ms 就花在连接建立或网络传输上问题不在业务逻辑。实际工程里我会给每条请求生成一个X-Trace-Id客户端发出去时带上服务端收到后原样记录这样即使并发请求也能精确配对。这个改动很小但排查效率提升明显import uuid # 客户端侧 trace_id str(uuid.uuid4()) conn.request(GET, path, headers{X-Trace-Id: trace_id}) # 服务端侧在 do_GET 里 trace_id self.headers.get(X-Trace-Id, unknown) entry[trace_id] trace_id有了 trace_id日志里grep一下就能把一次往返的两端记录捞出来不用靠时间戳猜。3. 客户端侧要控住的四个参数超时、复用、重试、畸形请求3.1 超时设置连接超时和读超时是两回事很多人只设一个timeout结果分不清是连不上还是连上了不返回。http.client的timeout参数同时作用于连接建立和每次 socket 读操作但实际排查时你需要区分。更精细的做法是用socket层控制或者换requests库分别设connect timeout和read timeoutimport requests # (connect_timeout, read_timeout) 元组形式 try: r requests.get(http://127.0.0.1:18080/slow, timeout(2, 10)) except requests.exceptions.ConnectTimeout: print(连接阶段超时服务端可能没起来或端口不通) except requests.exceptions.ReadTimeout: print(连接成功但读响应超时服务端处理太慢)参数说明(2, 10)表示连接最多等 2 秒读响应最多等 10 秒。这个区分在排查unexpected status 502 bad gateway这类错误时特别有用——502 通常是网关连不上后端属于连接阶段问题而不是读超时。3.2 连接复用keep-alive 到底复用了没有HTTP 连接复用是性能优化的常见手段但复用失败往往悄无声息。用http.client时同一个HTTPConnection对象连续发多次请求才会复用底层 socket每次conn.close()后再新建就是全新连接。验证方法是在服务端记录client_address的端口号复用成功时端口号不变conn http.client.HTTPConnection(127.0.0.1, 18080) for i in range(3): conn.request(GET, f/req{i}) resp conn.getresponse() resp.read() # 三次请求共用一条 TCP 连接服务端看到的源端口相同 conn.close()如果服务端日志里三次请求的源端口一致说明复用生效如果每次端口都变检查是否在循环里误建了新连接或者服务端返回了Connection: close头强制断开。3.3 重试策略不是所有失败都该重试重试能提高成功率但用错地方会放大故障。幂等的 GET 请求可以重试POST 创建订单重试可能导致重复下单。我一般只对连接类错误和 5xx 状态码重试且限制次数def get_with_retry(url, retries3): for i in range(retries): try: r requests.get(url, timeout(2, 5)) if r.status_code 500: return r # 4xx 不重试是客户端问题 except (requests.exceptions.ConnectionError, requests.exceptions.Timeout): if i retries - 1: raise time.sleep(0.5 * (i 1)) # 退避避免雪崩 return r参数说明retries3是总尝试次数0.5 * (i 1)是线性退避第一次等 0.5 秒第二次 1 秒。生产环境更常用指数退避加随机抖动避免多个客户端同时重试打垮服务端。3.4 构造畸形请求验证对端健壮性测试工具的价值之一是能发出正常客户端不会发的请求。比如超长 header、缺失 Host 头、Content-Length 与实际 body 不符。用原始 socket 可以绕过库的校验import socket # 发送 Content-Length 声明 100 但只发 10 字节 raw ( POST /upload HTTP/1.1\r\n Host: 127.0.0.1:18080\r\n Content-Length: 100\r\n \r\n 0123456789 ) s socket.create_connection((127.0.0.1, 18080), timeout5) s.sendall(raw.encode()) # 观察服务端是等待剩余 90 字节、超时关闭、还是直接报错 print(s.recv(1024)) s.close()这类测试用来验证服务端的超时处理和边界校验。如果服务端一直等剩余字节不释放连接就是资源泄漏隐患如果直接返回 400说明校验到位。4. 服务端侧要模拟的五种异常让对端在真实故障下暴露问题4.1 返回指定状态码和自定义响应头服务端测试工具的核心能力是「想返回什么就返回什么」。除了 200还要能返回 502、504、429 这些真实网关会产生的状态码验证客户端是否正确处理class FaultHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): # 根据路径决定返回什么故障 if self.path /502: self.send_response(502) self.send_header(Content-Type, text/plain) self.end_headers() self.wfile.write(bBad Gateway) elif self.path /429: self.send_response(429) self.send_header(Retry-After, 5) self.end_headers() else: self.send_response(200) self.end_headers()参数说明Retry-After: 5告诉客户端 5 秒后重试用来验证客户端是否遵守这个头。很多客户端忽略它直接立即重试这就是需要暴露的问题。4.2 延迟响应与提前断连延迟用来测试客户端超时设置是否合理提前断连用来测试客户端对不完整响应的处理import time class SlowHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): if self.path /slow: time.sleep(3) # 延迟 3 秒再响应 elif self.path /abort: # 发送部分响应头后直接关闭连接 self.wfile.write(bHTTP/1.1 200 OK\r\n) self.wfile.flush() self.connection.close() return self.send_response(200) self.end_headers()/abort路径模拟的是服务端处理到一半崩溃的场景。客户端如果只检查状态码不检查 body 完整性就会把不完整数据当成功这是隐蔽的 bug。4.3 大 body 与 chunked 传输测试客户端对大响应的处理以及是否正确解析 chunked 编码class ChunkedHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): self.send_response(200) self.send_header(Transfer-Encoding, chunked) self.end_headers() for i in range(5): chunk fchunk-{i}\n.encode() self.wfile.write(f{len(chunk):X}\r\n.encode()) self.wfile.write(chunk) self.wfile.write(b\r\n) self.wfile.write(b0\r\n\r\n) # 结束块chunked 编码的格式是「十六进制长度 CRLF 数据 CRLF」最后以 0 长度块结束。客户端解析错误会导致数据截断或卡住这个 handler 能稳定复现。4.4 并发连接与连接数限制验证客户端连接池和服务端并发处理能力需要服务端能同时接受多条连接并记录并发数import socketserver class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): daemon_threads True # 每个连接一个线程记录当前活跃连接数 active 0用ThreadingMixIn让每个连接独立线程处理配合计数器就能观察并发峰值。客户端侧用线程池发 50 个并发请求看服务端是否全部正确响应、有没有连接被拒绝。5. 避坑与排查这类工具最容易翻车的五个地方5.1 端口被占用导致服务端起不来现象启动时报OSError: [Errno 98] Address already in use或者客户端连上去返回的不是自己预期的响应。原因18080 端口被上一个没退干净的进程占着或者被其他服务用了。守护线程模式下主程序退出但 socket 没释放的情况也常见。解决启动前先探测端口或者用SO_REUSEADDR允许地址复用。更稳妥的是让操作系统分配随机端口启动后打印实际端口srv http.server.HTTPServer((127.0.0.1, 0), Handler) actual_port srv.server_address[1] print(fserver on port {actual_port})5.2 客户端读响应卡死现象客户端发出请求后一直不返回程序挂住。原因服务端声明了Content-Length但实际写入的 body 长度不符客户端按声明长度读读不满就一直等。或者服务端没发Content-Length也没用 chunked客户端不知道何时结束。解决服务端要么正确设置Content-Length要么用 chunked 编码要么在响应后关闭连接让客户端靠 EOF 判断结束。排查时用curl -v看响应头是否完整。5.3 连接复用导致的请求串味现象第二个请求收到了第一个请求的响应或者响应内容对不上。原因HTTP/1.1 的 keep-alive 连接上响应必须按请求顺序返回。如果服务端多线程处理但没保证顺序或者客户端没读完上一个响应就发下一个就会串。解决客户端必须完整读取上一个响应包括 body再发下一个请求。服务端如果用多线程要确保同一连接上的响应串行化。排查时给每个请求加唯一 trace_id看响应里的 trace_id 是否匹配。5.4 超时设置过短导致误判现象本地测试全通过一上测试环境就大量超时。原因本地回环延迟接近 0超时设 1 秒够用跨机房或经过网关后延迟可能几百毫秒加上服务端处理时间就超了。解决超时值要基于真实链路测量不要拍脑袋。连接超时和读超时分开设读超时给业务处理留足余量。用工具记录每次请求的实际耗时分布取 P99 再加缓冲。5.5 畸形请求测试影响正常用例现象跑完畸形请求测试后后续正常测试也开始失败。原因畸形请求可能让服务端进入异常状态比如半开连接没清理、线程池被占满。或者测试用的端口没释放正常用例连到了残留的故障 handler 上。解决畸形测试用独立的服务端实例和端口跑完立即销毁。每个测试用例之间做状态清理不要共用长生命周期的服务端。6. 进阶把两端观测数据变成可回放的测试用例工具跑通之后真正提升效率的一步是把每次往返的完整数据落盘变成可回放的用例。我一般会记录请求方法、路径、请求头、请求 body、响应状态码、响应头、响应 body、耗时、trace_id存成 JSON 行格式一行一条import json def record_case(trace_id, req, resp, cost): case { trace_id: trace_id, request: { method: req.method, path: req.path, headers: dict(req.headers), body: req.body.decode() if req.body else None, }, response: { status: resp.status, headers: dict(resp.headers), body: resp.body.decode() if resp.body else None, }, cost_ms: round(cost, 2), } with open(cases.jsonl, a) as f: f.write(json.dumps(case, ensure_asciiFalse) \n)有了这个文件回放就很简单读一行用请求部分重发对比响应部分是否一致。不一致的地方就是回归点。这个机制在接口重构时特别有用——改完代码跑一遍回放哪些接口行为变了立刻暴露。回放时要注意几个细节。响应里的时间戳、随机 ID 这类字段每次不同对比前要过滤掉。耗时字段只做参考不做断言因为环境不同耗时必然有差异。状态码和业务字段是强断言必须一致。再进一步可以把这些用例按 trace_id 分组标记哪些是正常路径、哪些是故障注入形成一个小型回归集。每次改完服务端或客户端代码跑一遍这个集子比手写单元测试覆盖更真实。我自己的习惯是每排查完一个线上问题就把复现它的那次往返记录留下来加进回归集。这样踩过的坑不会踩第二次。工具本身不复杂难的是坚持把每次故障变成可回放的资产。希望帮到你。本文还有配套的精品资源点击获取