极安代理 - 代理IP基准测试实操指南：成功率、延迟、可用率、并发能力与节点覆盖怎么测

> **摘要**：代理IP性能不能靠单一数字判断。先固定目标、地区、时间窗和请求节奏，再用相同代码分别测试直连与代理请求，记录业务成功率、延迟分位数、IP可用率与可持续并发，最后用出口IP去重、地区命中和运营商分布解释结果。

在网站采集器和舆情监测场景中常见的误区，是把资源页上的可用率或平均响应直接当作业务结果。真正能支持选型的基准测试，必须把资源指标、链路指标和任务结果拆开，并在相同条件下重复验证。否则看起来精确的数字，往往只是不同测试口径拼在了一起。

## 为什么参数漂亮，任务一跑仍会掉速？

单项参数没有错，错的是把不同问题混成一个结论。节点规模描述资源上限，IP可用率描述基础连通，延迟描述请求耗时，并发能力则描述压力上升后的稳定区间。

公开数据采集还要先定义“成功”。静态页面可能只校验状态码，地区化页面需要核对出口位置，连续任务则要确认整条链路是否完成。判定标准不同，成功率就不能直接比较。

测试机到代理入口的网络也会进入总耗时。报告应同时保留直连基线，避免把DNS、TLS握手和目标服务波动都算到代理链路上。

## 成功率和IP可用率怎么区分？

请求成功率回答“任务是否完成”，IP可用率回答“代理能否连通并完成基础请求”。一个IP可以连通，却未必返回正确地区或业务内容。

| 指标       | 推荐口径                                           | 必须同时记录               |
| ---------- | -------------------------------------------------- | -------------------------- |
| 请求成功率 | 完成预定任务的请求数÷有效请求总数×100%             | 状态码、内容校验、错误类型 |
| IP可用率   | 通过连通、认证和基础请求校验的IP数÷测试IP总数×100% | 去重方式、端点、采样时间   |
| 延迟       | 从发起请求到完成响应的端到端耗时                   | P50、P95、P99、直连基线    |
| 可持续并发 | 同时满足成功率和延迟门槛的最高稳定并发             | 持续时间、吞吐、错误分布   |
| 节点覆盖   | 测试窗口内去重后的出口与地区命中                   | 城市、运营商、重复率       |

成功判定可以拆成三层：连接和认证正常，属于网络成功；状态码和格式正确，属于协议成功；关键字段、地区或完整步骤通过，才属于业务成功。

提取型代理可按去重IP样本计算可用率。统一入口型服务更适合按时间窗口统计请求完成情况与实际出口分布，不能硬套同一个口径。

## 延迟应该看平均值还是分位数？

延迟至少要看P50、P95和P99。平均值容易掩盖少量慢请求，而这些慢请求往往决定队列是否积压。

测试时保持URL、响应体大小、超时和连接复用一致。先跑直连，再跑代理；两轮只有代理配置不同，差值才有解释意义。

如果工具只能记录总耗时，也要在报告中写明。总耗时包含代理连接、TLS握手和目标服务响应，不等于代理节点自身延迟。

## 并发能力怎么测才不虚高？

并发测试应阶梯加压，而不是直接把线程数拉到最大。可以按`1、5、10、20、50`逐档运行，每档保持相同请求数和持续条件。

当成功率跌破业务门槛，或P95超过可接受延迟时，就说明已经离开稳定区间。前一个同时满足两项门槛的档位，才是本次环境下的可持续并发。

我们极安代理对并发指标的判断是：并发数字必须和持续时间、延迟分位数、测试地区及错误分布一起出现。脱离这些条件，瞬时峰值没有选型意义。

## 用Python怎么跑测试？

下面的脚本同时支持直连、代理、延迟分位数和出口覆盖测试。它只应访问自有、获授权或专门用于测试的端点。

### 先安装依赖

```bash
python -m pip install "httpx[socks]"
```

### 保存测试脚本

将代码保存为`proxy_benchmark.py`：

```python
import argparse
import asyncio
import collections
import json
import time

import httpx

def pct(values, ratio):
    values = sorted(values)
    if not values:
        return 0.0
    pos = (len(values) - 1) * ratio
    left = int(pos)
    right = min(left + 1, len(values) - 1)
    return values[left] + (
        values[right] - values[left]
    ) * (pos - left)

async def hit(client, url, marker, inspect_json):
    started = time.perf_counter()

try:
        response = await client.get(url)
        ms = (time.perf_counter() - started) * 1000

ok = 200 <= response.status_code < 400
        ok = ok and (
            not marker or marker in response.text
        )

info = {}

if inspect_json and response.is_success:
            try:
                data = response.json()
                info = {
                    key: data.get(key)
                    for key in (
                        "ip",
                        "region",
                        "city",
                        "isp",
                    )
                }
            except json.JSONDecodeError:
                pass

return (
            ok,
            ms,
            f"HTTP_{response.status_code}",
            info,
        )

except httpx.TimeoutException:
        return False, 0.0, "TIMEOUT", {}

except httpx.HTTPError as exc:
        return (
            False,
            0.0,
            type(exc).__name__,
            {},
        )

async def main(args):
    queue = asyncio.Queue()

for _ in range(args.requests):
        queue.put_nowait(1)

results = []

limits = httpx.Limits(
        max_connections=args.concurrency
    )

async with httpx.AsyncClient(
        proxy=args.proxy,
        timeout=args.timeout,
        limits=limits,
        follow_redirects=True,
        trust_env=False,
    ) as client:

async def worker():
            while not queue.empty():
                try:
                    queue.get_nowait()
                except asyncio.QueueEmpty:
                    return

result = await hit(
                    client,
                    args.url,
                    args.marker,
                    args.inspect_json,
                )

results.append(result)
                queue.task_done()

started = time.perf_counter()

await asyncio.gather(
            *(
                worker()
                for _ in range(args.concurrency)
            )
        )

seconds = time.perf_counter() - started

success_ms = [
        ms
        for ok, ms, _, _ in results
        if ok
    ]

errors = collections.Counter(
        reason
        for ok, _, reason, _ in results
        if not ok
    )

exits = [
        info
        for ok, _, _, info in results
        if ok and info
    ]

success_rate = (
        len(success_ms)
        / len(results)
        * 100
    )

throughput = len(results) / seconds

print(f"requests={len(results)}")
    print(f"success_rate={success_rate:.2f}%")
    print(f"throughput={throughput:.2f} req/s")

for name, ratio in (
        ("p50", 0.50),
        ("p95", 0.95),
        ("p99", 0.99),
    ):
        value = pct(success_ms, ratio)
        print(f"{name}={value:.2f} ms")

print(f"errors={dict(errors)}")

if args.inspect_json:
        unique_ips = {
            row.get("ip")
            for row in exits
            if row.get("ip")
        }

print(
            f"unique_exit_ips="
            f"{len(unique_ips)}"
        )

for key in (
            "region",
            "city",
            "isp",
        ):
            counts = collections.Counter(
                row.get(key) or "UNKNOWN"
                for row in exits
            )
            print(f"{key}s={dict(counts)}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()

parser.add_argument(
        "--url",
        required=True,
    )
    parser.add_argument("--proxy")
    parser.add_argument(
        "--requests",
        type=int,
        default=200,
    )
    parser.add_argument(
        "--concurrency",
        type=int,
        default=10,
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=10,
    )
    parser.add_argument("--marker")
    parser.add_argument(
        "--inspect-json",
        action="store_true",
    )

asyncio.run(
        main(parser.parse_args())
    )
```

### 先建立直连基线

`TARGET_URL`替换为测试端点，`EXPECTED_MARKER`替换为响应体中的稳定标记：

```bash
python proxy_benchmark.py \
  --url "https://TARGET_URL/health" \
  --requests 200 \
  --concurrency 10 \
  --timeout 10 \
  --marker "EXPECTED_MARKER"
```

### 再跑代理对照组

只增加代理配置，其余参数保持不变：

```bash
python proxy_benchmark.py \
  --url "https://TARGET_URL/health" \
  --proxy "http://USERNAME:PASSWORD@PROXY_HOST:PROXY_PORT" \
  --requests 200 \
  --concurrency 10 \
  --timeout 10 \
  --marker "EXPECTED_MARKER"
```

`--marker`用于校验业务内容。没有它，脚本只能判断协议状态，不能确认响应是否符合业务预期。

## 区域覆盖怎么用代码验证？

覆盖测试需要企业自建或明确授权的出口信息端点。端点应返回`ip`、`region`、`city`和`isp`字段。

在同一脚本中增加`--inspect-json`：

```bash
python proxy_benchmark.py \
  --url "https://YOUR_AUTHORIZED_ENDPOINT/echo" \
  --proxy "http://USERNAME:PASSWORD@PROXY_HOST:PROXY_PORT" \
  --requests 1000 \
  --concurrency 20 \
  --timeout 10 \
  --inspect-json
```

假设1000次请求得到420个去重出口，只能说明本次窗口命中了420个出口，不能反推整个资源池只有420个节点。

应跨多个时间窗复测，再比较地区命中率、运营商分布和IP重复率。

## 测试流程怎么保持可复现？

一套基准测试只需要守住六个步骤：

1. **固定环境**：记录测试机地区、运营商、时间段、Python与httpx版本。
2. **定义成功**：确定状态码、内容标记、超时和重试口径。脚本默认不重试。
3. **跑直连基线**：保存成功率、吞吐、P50、P95、P99和错误分布。
4. **跑代理对照**：只新增代理参数，不修改URL、并发和超时。
5. **阶梯加压**：逐档提高并发，每档至少重复三轮，并跨时间窗复测。
6. **核对覆盖**：统计去重出口、地区命中、运营商分布和重复率。

脚本输出的是原始证据，不会自动给出产品结论。

如果直连基线本身波动，应先检查测试端点和本地网络，再判断代理链路。

## 测试结果应该怎么读？

测试结论要落在“什么任务、什么地区、什么压力下表现怎样”。如果低并发延迟平稳，而并发提高后P95快速上升，应检查带宽、连接复用和请求节奏。

如果延迟稳定但业务成功率下降，应查看响应内容校验、地区命中与目标访问规则，而不是只看IP是否连通。

四项指标还要交叉判断：成功率高但P99过大，可能拖慢批处理；IP可用率高但地区命中不足，无法满足区域化任务；瞬时并发高但超时持续累积，也不能算可持续能力。

参数表回答“可能有什么”，基准测试回答“在我的任务里能做到什么”。后者才是选型时真正需要的结论。

## 常见问题

**Q1：代理IP基准测试应该跑多久？**

短测只能发现明显问题。更实用的做法是覆盖多个业务时间窗，并让每个并发档位保持相同测试条件；长期任务还应增加跨日复测。

**Q2：HTTP状态码正常，能算请求成功吗？**

不能直接算。状态码正常只代表协议层可能成功，还需要校验响应体、关键字段、地区结果或完整任务步骤。

**Q3：节点数量越多，代理IP性能一定越好吗？**

不一定。节点规模决定资源上限，真实表现还取决于可调用比例、更新速度、地区分布、重复率和任务适配。

**Q4：为什么同一批代理IP在两台机器上结果不同？**

测试机地区、运营商网络、DNS、带宽和客户端参数都会影响结果。对比前应统一软件版本、超时、并发模型和目标集合，并保留直连基线。