OpenCode 实践课:4 分钟用 Python 写出一个命令行 HTTP 接口压测工具

项目介绍

后端接口上线前,我们经常需要一个简单工具来验证接口的并发承受能力。虽然市面上有 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

准备工作

  • Python:3.8 及以上(预装 argparse)
  • OpenCode:安装与配置见 opencode.ai 官方文档
  • 零依赖:只用 Python 标准库,不需要 pip install 任何包

打开终端,进入项目目录后启动 OpenCode:

mkdir httpbench && cd httpbench
opencode

进入交互界面后,接下来全程用自然语言驱动开发。

实践过程

第一步:让 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 负责将意图转化为可运行的代码。对于日常的小工具、原型验证、脚本开发这类场景,效率提升是数量级的。