脚本如何统计接口响应时间

wen 实用脚本 2026-06-30 26

本文目录导读：

脚本如何统计接口响应时间

使用 curl 命令（Linux/Mac）
Python 脚本统计
Shell 脚本循环测试
Node.js 脚本
Go 语言脚本
专业工具
完整统计脚本（Python）
注意事项

使用 curl 命令（Linux/Mac）

# 使用 time 命令
time curl -s -o /dev/null -w "响应时间: %{time_total}s\n" https://api.example.com/endpoint
# 或更详细的统计
curl -s -o /dev/null -w "
连接时间: %{time_connect}s
开始传输: %{time_starttransfer}s
总时间: %{time_total}s
速度: %{speed_download} B/s\n" https://api.example.com/endpoint

Python 脚本统计

使用 requests 库

import requests
import time
import statistics
def measure_response_time(url, method='GET', **kwargs):
    """统计接口响应时间"""
    # 单次请求
    start = time.time()
    if method == 'GET':
        response = requests.get(url, **kwargs)
    elif method == 'POST':
        response = requests.post(url, **kwargs)
    end = time.time()
    response_time = (end - start) * 1000  # 转换为毫秒
    return {
        'status_code': response.status_code,
        'response_time_ms': round(response_time, 2),
        'response_size': len(response.content)
    }
# 多次请求统计
def batch_test(url, times=10):
    """多次测试获取统计信息"""
    times_list = []
    for i in range(times):
        result = measure_response_time(url)
        times_list.append(result['response_time_ms'])
        print(f"请求 {i+1}: {result['response_time_ms']}ms")
    # 统计信息
    stats = {
        '平均响应时间': round(statistics.mean(times_list), 2),
        '最大响应时间': round(max(times_list), 2),
        '最小响应时间': round(min(times_list), 2),
        '中位数': round(statistics.median(times_list), 2),
        '标准差': round(statistics.stdev(times_list), 2) if len(times_list) > 1 else 0
    }
    return stats
# 使用示例
url = "https://api.example.com/endpoint"
stats = batch_test(url, 5)
print("\n统计结果：")
for key, value in stats.items():
    print(f"{key}: {value}ms")

使用 aiohttp（异步版本）

import asyncio
import aiohttp
import time
async def measure_async(url, session):
    start = time.time()
    async with session.get(url) as response:
        await response.text()
    end = time.time()
    return (end - start) * 1000
async def batch_test_async(url, times=10):
    connector = aiohttp.TCPConnector(limit=10)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [measure_async(url, session) for _ in range(times)]
        times_list = await asyncio.gather(*tasks)
    print("异步测试结果：")
    for i, t in enumerate(times_list):
        print(f"请求 {i+1}: {t:.2f}ms")
    return times_list
# 运行异步测试
# asyncio.run(batch_test_async("https://api.example.com/endpoint"))

Shell 脚本循环测试

#!/bin/bash
url="https://api.example.com/endpoint"
times=10
total_time=0
min_time=999999
max_time=0
for ((i=1; i<=times; i++))
do
    # 记录开始时间
    start=$(date +%s%N)
    # 发送请求
    response=$(curl -s -o /dev/null -w "%{http_code}" $url)
    # 记录结束时间
    end=$(date +%s%N)
    # 计算耗时（毫秒）
    duration=$(( ($end - $start) / 1000000 ))
    echo "请求 $i: 状态码=$response, 耗时=${duration}ms"
    total_time=$((total_time + duration))
    if [ $duration -lt $min_time ]; then
        min_time=$duration
    fi
    if [ $duration -gt $max_time ]; then
        max_time=$duration
    fi
done
avg_time=$((total_time / times))
echo "--------------------"
echo "统计结果 (共${times}次请求):"
echo "平均响应时间: ${avg_time}ms"
echo "最小响应时间: ${min_time}ms"
echo "最大响应时间: ${max_time}ms"

Node.js 脚本

const axios = require('axios');
async function measureResponseTime(url, times = 10) {
    const timesList = [];
    for (let i = 0; i < times; i++) {
        const start = Date.now();
        try {
            const response = await axios.get(url);
            const end = Date.now();
            const duration = end - start;
            timesList.push(duration);
            console.log(`请求 ${i+1}: ${duration}ms, 状态码: ${response.status}`);
        } catch (error) {
            console.log(`请求 ${i+1}: 失败 - ${error.message}`);
        }
    }
    if (timesList.length > 0) {
        const avg = timesList.reduce((a, b) => a + b, 0) / timesList.length;
        const max = Math.max(...timesList);
        const min = Math.min(...timesList);
        console.log('\n统计结果:');
        console.log(`平均响应时间: ${avg.toFixed(2)}ms`);
        console.log(`最大响应时间: ${max}ms`);
        console.log(`最小响应时间: ${min}ms`);
    }
}
// 使用示例
measureResponseTime('https://api.example.com/endpoint', 5);

Go 语言脚本

package main
import (
    "fmt"
    "net/http"
    "time"
    "math"
)
func measureResponseTime(url string) (time.Duration, error) {
    start := time.Now()
    resp, err := http.Get(url)
    if err != nil {
        return 0, err
    }
    defer resp.Body.Close()
    elapsed := time.Since(start)
    return elapsed, nil
}
func main() {
    url := "https://api.example.com/endpoint"
    times := 10
    var total time.Duration
    var min, max time.Duration
    for i := 0; i < times; i++ {
        duration, err := measureResponseTime(url)
        if err != nil {
            fmt.Printf("请求 %d: 失败 - %v\n", i+1, err)
            continue
        }
        fmt.Printf("请求 %d: %v\n", i+1, duration)
        total += duration
        if i == 0 || duration < min {
            min = duration
        }
        if duration > max {
            max = duration
        }
        time.Sleep(100 * time.Millisecond) // 避免请求过快
    }
    avg := time.Duration(int64(total) / int64(times))
    fmt.Printf("\n统计结果:\n")
    fmt.Printf("平均响应时间: %v\n", avg)
    fmt.Printf("最小响应时间: %v\n", min)
    fmt.Printf("最大响应时间: %v\n", max)
}

专业工具

Apache Bench (ab)

# 基本用法
ab -n 100 -c 10 https://api.example.com/endpoint
# 带 POST 请求
ab -n 100 -c 10 -p post_data.json -T application/json https://api.example.com/endpoint

wrk

wrk -t12 -c100 -d30s https://api.example.com/endpoint

httperf

httperf --server api.example.com --port 443 --uri /endpoint --num-conns 100 --rate 10

完整统计脚本（Python）

#!/usr/bin/env python3
"""
接口响应时间统计工具
"""
import requests
import time
import statistics
import sys
from datetime import datetime
class APITester:
    def __init__(self, url, method='GET', headers=None, data=None, timeout=30):
        self.url = url
        self.method = method
        self.headers = headers or {}
        self.data = data
        self.timeout = timeout
        self.results = []
    def single_test(self):
        """单次测试"""
        try:
            start = time.time()
            if self.method == 'GET':
                response = requests.get(
                    self.url,
                    headers=self.headers,
                    timeout=self.timeout
                )
            elif self.method == 'POST':
                response = requests.post(
                    self.url,
                    headers=self.headers,
                    data=self.data,
                    timeout=self.timeout
                )
            end = time.time()
            response_time = (end - start) * 1000
            return {
                'timestamp': datetime.now().isoformat(),
                'response_time': round(response_time, 2),
                'status_code': response.status_code,
                'response_size': len(response.content)
            }
        except requests.exceptions.RequestException as e:
            return {
                'timestamp': datetime.now().isoformat(),
                'response_time': None,
                'error': str(e)
            }
    def batch_test(self, times=10, interval=0):
        """批量测试"""
        print(f"开始测试: {self.url}")
        print(f"请求方法: {self.method}")
        print(f"测试次数: {times}")
        print("-" * 40)
        for i in range(times):
            result = self.single_test()
            self.results.append(result)
            if result['response_time']:
                print(f"请求 {i+1}: {result['response_time']}ms " 
                      f"(状态码: {result['status_code']})")
            else:
                print(f"请求 {i+1}: 失败 - {result.get('error', '未知错误')}")
            if i < times - 1 and interval > 0:
                time.sleep(interval)
        # 分析结果
        self.analyze_results()
    def analyze_results(self):
        """分析测试结果"""
        successful = [r for r in self.results if r['response_time'] is not None]
        failed = [r for r in self.results if r['response_time'] is None]
        if not successful:
            print("\n所有请求均失败！")
            return
        times_list = [r['response_time'] for r in successful]
        print("\n" + "=" * 40)
        print("测试统计结果")
        print("=" * 40)
        print(f"总请求数: {len(self.results)}")
        print(f"成功: {len(successful)}")
        print(f"失败: {len(failed)}")
        print(f"成功率: {len(successful)/len(self.results)*100:.1f}%")
        print(f"平均响应时间: {statistics.mean(times_list):.2f}ms")
        print(f"中位数: {statistics.median(times_list):.2f}ms")
        print(f"最小响应时间: {min(times_list):.2f}ms")
        print(f"最大响应时间: {max(times_list):.2f}ms")
        if len(times_list) > 1:
            print(f"标准差: {statistics.stdev(times_list):.2f}ms")
        # 计算百分位
        sorted_times = sorted(times_list)
        print(f"90% 请求在: {sorted_times[int(len(sorted_times)*0.9)-1]:.2f}ms 以内")
        print(f"95% 请求在: {sorted_times[int(len(sorted_times)*0.95)-1]:.2f}ms 以内")
        print(f"99% 请求在: {sorted_times[int(len(sorted_times)*0.99)-1]:.2f}ms 以内")
# 使用示例
if __name__ == "__main__":
    # 基本用法
    tester = APITester(
        url="https://api.example.com/endpoint",
        method="GET",
        headers={"Authorization": "Bearer your_token"}
    )
    tester.batch_test(times=10, interval=0.5)

注意事项

网络环境：测试结果受网络状况影响，建议在稳定的网络环境下测试
请求频率：避免过高频率请求，防止被限流
超时设置：设置合理的超时时间
并发测试：使用专业工具（如 ab、wrk）进行并发测试
结果缓存：多次测试取平均值更准确
异常处理：处理好网络错误、超时等异常情况

选择哪种方法取决于你的具体需求：

快速测试：使用 curl 或 ab
详细统计：使用 Python 或 Node.js 脚本
性能测试：使用 wrk 或 Apache Bench