后端接口上线前,我们经常需要一个简单工具来验证接口的并发承受能力。虽然市面上有 ApacheBench、wrk 等专业工具,但它们需要额外安装,配置也不够灵活。今天我们用 OpenCode 纯自然语言指挥 AI,4 分钟打造一个零依赖的命令行 HTTP 压测工具。
技术栈:Python 3 + 标准库(http.client / threading / argparse)
最终效果:一个单文件 Python 脚本,通过命令行发起并发 HTTP 请求,输出 QPS、响应时间分布、状态码统计等核心指标。
$ python benchmark.py https://httpbin.org/get -n 50 -c 10
==================================================
目标: https://httpbin.org/get
==================================================
总请求: 50
并发数: 10
总耗时: 4321.5 ms
QPS: 11.57
成功: 50
失败: 0
==================================================
响应时间 (ms):
最小: 89.2
最大: 952.7
平均: 342.8
中位: 298.6
==================================================
状态码分布:
200: 50
pip install 任何包打开终端,进入项目目录后启动 OpenCode:
mkdir httpbench && cd httpbench opencode
进入交互界面后,接下来全程用自然语言驱动开发。
我对 OpenCode 说:
> 写一个 Python 命令行工具,用 argparse 接收一个 URL 参数、-n 请求数、-c 并发线程数。用 http.client 发 GET 请求,测量每条请求的耗时。把项目结构建好。
OpenCode 的响应:
OpenCode 自动创建了 benchmark.py,生成完整的参数解析逻辑,用 urllib.parse.urlparse 解析用户输入的 URL,自动判断 HTTP/HTTPS 协议、默认端口号。同时写好了单次请求函数 make_request(),用 time.perf_counter() 精确计时。
关键要点:OpenCode 能理解"并发"的意图,主动引入 threading 模块;对 URL 解析、协议判断这些细节也处理到位,不需要我单独说明。这就是 AI 编程的核心优势——你说的是"要什么",而非"怎么写"。
我对 OpenCode 说:
> 现在只有一个单次请求函数,帮我实现多线程并发。把总请求数均分给每个线程,线程间安全地收集结果。
OpenCode 的响应:
OpenCode 创建了 BenchmarkResult 类,用 threading.Lock() 保证线程安全地写入数据。然后写了 worker() 函数作为线程入口,在主函数里按并发数创建线程、均分请求量(用整除和取余处理不能整除的情况)。
base = args.requests // args.concurrency
remainder = args.requests % args.concurrency
for i in range(args.concurrency):
count = base + (1 if i < remainder else 0)
t = threading.Thread(target=worker, args=(...))
threads.append(t)
t.start()
关键要点:这个"均分 + 余数分配"的逻辑如果用传统方式需要自己写边界判断,但 OpenCode 一下就搞定了。当我问"如果请求数不能被线程数整除怎么办",它已经在代码里处理好了。
我对 OpenCode 说:
> 计算 QPS、最小/最大/平均/中位响应时间,统计 HTTP 状态码分布,用对齐格式漂亮地打印出来。失败的请求统计为"错误"。
OpenCode 的响应:
OpenCode 补齐了 main() 函数中的统计计算和打印逻辑。排序后取中位数,用 result.codes.get(0, 0) 统计异常请求数,输出格式统一用等号分隔线和对齐的标签。
关键要点:格式化输出这种"体力活"在传统开发中最花时间——对齐空格、排版布局、各种边界情况。OpenCode 一次对话就生成了工整的输出,我只用了一句描述。
我对 OpenCode 说:
> 用 httpbin.org 帮我写个测试命令验证工具是否正常工作。另外检查一下,如果目标服务完全不可达会不会崩溃?
OpenCode 的响应:
OpenCode 建议了测试命令:
python benchmark.py https://httpbin.org/get -n 20 -c 5
同时指出 make_request() 中的 try/except 已经处理了连接失败的场景,异常请求的 status 记为 0,不会导致程序崩溃。
import argparse
import http.client
import time
import threading
import urllib.parse
import sys
def parse_args():
parser = argparse.ArgumentParser(
description="HTTP 接口压测工具 — 用并发请求测试 API 性能",
epilog="示例: python benchmark.py https://httpbin.org/get -n 50 -c 10",
)
parser.add_argument("url", help="目标 URL")
parser.add_argument("-n", "--requests", type=int, default=10, help="总请求数,默认 10")
parser.add_argument("-c", "--concurrency", type=int, default=1, help="并发线程数,默认 1")
parser.add_argument("-t", "--timeout", type=int, default=30, help="请求超时(秒),默认 30")
return parser.parse_args()
def make_request(host, path, port, timeout, use_ssl):
conn_cls = http.client.HTTPSConnection if use_ssl else http.client.HTTPConnection
conn = conn_cls(host, port, timeout=timeout)
start = time.perf_counter()
try:
conn.request("GET", path)
resp = conn.getresponse()
resp.read()
elapsed = (time.perf_counter() - start) * 1000
conn.close()
return resp.status, elapsed
except Exception:
elapsed = (time.perf_counter() - start) * 1000
conn.close()
return 0, elapsed
class BenchmarkResult:
def __init__(self):
self.lock = threading.Lock()
self.durations = []
self.codes = {}
def add(self, status, duration):
with self.lock:
self.durations.append(duration)
self.codes[status] = self.codes.get(status, 0) + 1
def worker(host, path, port, timeout, use_ssl, result, count):
for _ in range(count):
status, duration = make_request(host, path, port, timeout, use_ssl)
result.add(status, duration)
def main():
args = parse_args()
parsed = urllib.parse.urlparse(args.url)
host = parsed.hostname
path = parsed.path or "/"
if parsed.query:
path += "?" + parsed.query
port = parsed.port or (443 if parsed.scheme == "https" else 80)
use_ssl = parsed.scheme == "https"
result = BenchmarkResult()
threads = []
base = args.requests // args.concurrency
remainder = args.requests % args.concurrency
t_start = time.perf_counter()
for i in range(args.concurrency):
count = base + (1 if i < remainder else 0)
t = threading.Thread(
target=worker,
args=(host, path, port, args.timeout, use_ssl, result, count),
)
threads.append(t)
t.start()
for t in threads:
t.join()
total_time = (time.perf_counter() - t_start) * 1000
durations = result.durations
n = len(durations)
if n == 0:
print("没有完成任何请求。")
return
avg = sum(durations) / n
sorted_d = sorted(durations)
median = sorted_d[n // 2] if n > 1 else sorted_d[0]
errors = result.codes.get(0, 0)
print(f"\n{'=' * 50}")
print(f" 目标: {args.url}")
print(f"{'=' * 50}")
print(f" 总请求: {n}")
print(f" 并发数: {args.concurrency}")
print(f" 总耗时: {total_time:.1f} ms")
print(f" QPS: {n / (total_time / 1000):.2f}")
print(f" 成功: {n - errors}")
print(f" 失败: {errors}")
print(f"{'=' * 50}")
print(f" 响应时间 (ms):")
print(f" 最小: {min(durations):.1f}")
print(f" 最大: {max(durations):.1f}")
print(f" 平均: {avg:.1f}")
print(f" 中位: {median:.1f}")
print(f"{'=' * 50}")
print(f" 状态码分布:")
for code, cnt in sorted(result.codes.items()):
label = "错误" if code == 0 else code
print(f" {label}: {cnt}")
print()
if __name__ == "__main__":
main()
整个项目从想法到可运行,实际用了大约 4 分钟。我全程没有手写一行代码,只需要用自然语言描述"要什么"——建骨架、加并发、做统计、调格式。
对比传统开发方式:
| 环节 | 传统方式 | 用 OpenCode |
|------|---------|-------------|
| 项目骨架 + 参数解析 | ~8 分钟 | 30 秒 |
| HTTP 请求 + 计时逻辑 | ~5 分钟 | 20 秒 |
| 多线程并发 + 线程安全 | ~10 分钟 | 40 秒 |
| 统计计算 + 格式化输出 | ~8 分钟 | 30 秒 |
| 调试修正 | ~5 分钟 | 1 分钟 |
| 合计 | ~36 分钟 | ~4 分钟 |
OpenCode 的价值不在于"替人写代码",而在于将"实现细节"从脑力负担变成了对话式协作。你关注项目设计和高层逻辑,AI 负责将意图转化为可运行的代码。对于日常的小工具、原型验证、脚本开发这类场景,效率提升是数量级的。