如何用脚本批量超时API请求？

wen 实用脚本 2026-07-22 2

本文目录导读：

如何用脚本批量超时API请求？

Python 脚本（推荐）
Bash 脚本（配合 curl）
Node.js 脚本
PowerShell 脚本（Windows）
使用建议

我可以帮您编写批量超时API请求的脚本,以下是几种常见语言的实现方案：

Python 脚本（推荐）

使用 asyncio + aiohttp（异步高并发）

import asyncio
import aiohttp
import time
from typing import List, Dict
async def fetch_with_timeout(session, url, timeout=5):
    """单个请求带超时"""
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as response:
            return {
                "url": url,
                "status": response.status,
                "time": time.time()
            }
    except asyncio.TimeoutError:
        return {
            "url": url,
            "status": "TIMEOUT",
            "time": time.time()
        }
    except Exception as e:
        return {
            "url": url,
            "status": f"ERROR: {str(e)}",
            "time": time.time()
        }
async def batch_request(urls: List[str], timeout=5, concurrency=10):
    """批量请求控制并发"""
    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [fetch_with_timeout(session, url, timeout) for url in urls]
        results = await asyncio.gather(*tasks)
    return results
# 使用示例
urls = [
    "https://api.example.com/endpoint1",
    "https://api.example.com/endpoint2",
    # ... 更多URL
]
results = asyncio.run(batch_request(urls, timeout=3, concurrency=20))
for r in results:
    print(f"{r['url']}: {r['status']} (时间: {r['time']})")

使用 requests + concurrent.futures（简单同步方式）

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict
def request_with_timeout(url, timeout=5):
    """单个同步请求带超时"""
    try:
        response = requests.get(url, timeout=timeout)
        return {"url": url, "status": response.status_code}
    except requests.Timeout:
        return {"url": url, "status": "TIMEOUT"}
    except Exception as e:
        return {"url": url, "status": f"ERROR: {str(e)}"}
def batch_requests_sync(urls: List[str], timeout=5, workers=10):
    """同步批量请求"""
    results = []
    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = {executor.submit(request_with_timeout, url, timeout): url 
                  for url in urls}
        for future in as_completed(futures):
            results.append(future.result())
    return results
# 使用示例
urls = ["https://api1.com", "https://api2.com", "https://api3.com"]
results = batch_requests_sync(urls, timeout=3, workers=5)

Bash 脚本（配合 curl）

#!/bin/bash
# 批量测试API超时脚本
API_URLS=(
    "https://api.example.com/endpoint1"
    "https://api.example.com/endpoint2"
    "https://api.example.com/endpoint3"
)
TIMEOUT=5  # 超时时间（秒）
LOG_FILE="api_timeout_test.log"
echo "开始批量API超时测试 - $(date)" | tee -a $LOG_FILE
for url in "${API_URLS[@]}"; do
    echo "测试: $url" | tee -a $LOG_FILE
    # 使用 curl 进行测试
    start_time=$(date +%s.%N)
    response=$(curl -s -o /dev/null -w "%{http_code}" \
        --connect-timeout $TIMEOUT \
        --max-time $(($TIMEOUT + 2)) \
        "$url" 2>&1)
    end_time=$(date +%s.%N)
    duration=$(echo "$end_time - $start_time" | bc)
    if [ $? -eq 0 ]; then
        echo "状态: $response | 耗时: ${duration}s" | tee -a $LOG_FILE
    else
        echo "状态: TIMEOUT/ERROR | 耗时: ${duration}s" | tee -a $LOG_FILE
    fi
    echo "---" | tee -a $LOG_FILE
done
echo "测试完成 - $(date)" | tee -a $LOG_FILE

Node.js 脚本

const axios = require('axios');
async function fetchWithTimeout(url, timeoutMs = 5000) {
    try {
        const response = await axios.get(url, {
            timeout: timeoutMs,
            validateStatus: false
        });
        return { url, status: response.status, success: true };
    } catch (error) {
        if (error.code === 'ECONNABORTED') {
            return { url, status: 'TIMEOUT', success: false };
        }
        return { url, status: `ERROR: ${error.message}`, success: false };
    }
}
async function batchRequests(urls, timeout = 5000, concurrency = 10) {
    const results = [];
    // 分批处理控制并发
    for (let i = 0; i < urls.length; i += concurrency) {
        const batch = urls.slice(i, i + concurrency);
        const promises = batch.map(url => fetchWithTimeout(url, timeout));
        const batchResults = await Promise.all(promises);
        results.push(...batchResults);
        // 可选：批次间延时
        if (i + concurrency < urls.length) {
            await new Promise(resolve => setTimeout(resolve, 100));
        }
    }
    return results;
}
// 使用示例
const urls = [
    'https://api.example.com/endpoint1',
    'https://api.example.com/endpoint2',
];
batchRequests(urls, 3000, 5).then(results => {
    results.forEach(r => {
        console.log(`${r.url}: ${r.status}`);
    });
});

PowerShell 脚本（Windows）

# PowerShell 批量API测试脚本
$urls = @(
    "https://api.example.com/endpoint1",
    "https://api.example.com/endpoint2",
    "https://api.example.com/endpoint3"
)
$timeoutSeconds = 5
$results = @()
foreach ($url in $urls) {
    Write-Host "Testing: $url" -ForegroundColor Yellow
    try {
        $stopwatch = [System.Diagnostics.Stopwatch]::StartNew()
        $response = Invoke-WebRequest -Uri $url -TimeoutSec $timeoutSeconds -ErrorAction Stop
        $stopwatch.Stop()
        $result = [PSCustomObject]@{
            URL = $url
            Status = $response.StatusCode
            Duration = $stopwatch.Elapsed.TotalSeconds
            Result = "Success"
        }
    } catch [System.Net.WebException] {
        $result = [PSCustomObject]@{
            URL = $url
            Status = "TIMEOUT/ERROR"
            Duration = $timeoutSeconds
            Result = $_.Exception.Message
        }
    }
    $results += $result
    Write-Host "  Result: $($result.Status) ($($result.Duration)s)" -ForegroundColor Cyan
}
# 输出结果
$results | Format-Table -AutoSize
$results | Export-Csv -Path "API_Timeout_Results.csv" -NoTypeInformation

使用建议

超时时间设置：根据API响应速度合理设置，通常3-10秒
并发控制：避免过高的并发导致目标服务器过载
错误处理：区分网络超时、服务器错误、连接拒绝等情况
日志记录：保存测试结果便于分析
重试机制：对于失败的请求可以添加重试逻辑

选择哪种方案取决于您的具体需求：Python适合复杂逻辑，Bash适合简单测试，Node.js适合全栈项目。