每分钟5万次POST请求的4小时压测问题及替代方案咨询

你的脚本出现超时和无法完成任务的核心原因有几个：

• requests是同步阻塞库，每个线程发起请求后会等待响应，默认线程池的并发数远不足以支撑每分钟5万次请求的吞吐量。
• 一次性提交5万任务后等待全部完成，完全没做速率控制——如果单请求耗时超过12ms（5万/60秒≈833次/秒），1分钟内根本完不成5万次，后续循环会不断叠加任务，导致任务队列越积越多，永远无法结束。
• 每个请求都打印响应内容，大量IO操作进一步拖慢执行效率。

方案1：用异步请求库+aiohttp做速率控制 #

异步非阻塞模型能在单进程内处理数千并发请求，配合速率控制可以精准达到每分钟5万次的目标。

示例代码：

import aiohttp
import asyncio
from datetime import datetime

async def send_request(session, url, headers, payload):
    try:
        async with session.post(url, headers=headers, json=payload):
            # 不需要打印每个响应，避免IO开销，如需统计可维护计数器
            pass
    except Exception as e:
        # 异常可批量记录，不要实时打印
        pass

async def main():
    url = "你的目标URL"
    headers = {"Content-Type": "application/json"}
    payload = {"固定的payload内容": "xxx"}
    
    events_per_minute = 50000
    events_per_second = events_per_minute / 60
    test_duration_hours = 4
    total_seconds = test_duration_hours * 3600

    # 控制并发数，根据目标服务性能调整，比如设为1000
    semaphore = asyncio.Semaphore(1000)
    
    async with aiohttp.ClientSession() as session:
        start_time = datetime.now()
        request_count = 0
        
        while (datetime.now() - start_time).total_seconds() < total_seconds:
            # 控制每秒发送的请求数，确保速率稳定
            batch_size = int(events_per_second)
            tasks = []
            for _ in range(batch_size):
                # 用信号量控制并发
                task = asyncio.create_task(
                    send_request(session, url, headers, payload)
                )
                tasks.append(task)
            await asyncio.gather(*tasks)
            request_count += batch_size
            
            # 调整间隔，确保速率精准
            elapsed = (datetime.now() - start_time).total_seconds()
            expected_requests = elapsed * events_per_second
            if request_count > expected_requests:
                await asyncio.sleep(0.1)
    
    print(f"测试完成，总请求数：{request_count}")

if __name__ == "__main__":
    asyncio.run(main())

方案2：使用专业负载测试工具 #

自己写脚本应对高并发负载测试很容易踩坑，专业工具自带成熟的速率控制、并发管理、统计分析功能，推荐两个：

Locust（Python编写，易定制） #

• 核心优势：用Python定义测试逻辑，支持分布式压测，能直观看到QPS、响应时间、错误率等指标。
• 简单示例（locustfile.py）：

from locust import HttpUser, task, between

class LoadTestUser(HttpUser):
    wait_time = between(0, 0)  # 不添加随机等待，严格控制速率
    target_qps = 50000 / 60  # 每秒请求数

    @task
    def send_post(self):
        self.client.post("/你的接口路径", json={"固定payload": "xxx"}, headers={"Content-Type": "application/json"})

    def on_start(self):
        # 可初始化配置
        pass

启动命令：locust -f locustfile.py --headless -u 1000 -r 833 -t 4h

• -u：并发用户数
• -r：每秒新增用户数（这里设为833，刚好每分钟5万）
• -t：测试时长

k6（Go编写，性能极高） #

• 核心优势：性能比Python工具更强，适合超高并发场景，用JS定义测试脚本。
• 简单示例（script.js）：

import http from 'k6/http';
import { sleep } from 'k6';

export const options = {
    vus: 1000,  // 虚拟用户数
    duration: '4h',  // 测试时长
    rps: 833,  // 每秒请求数，对应每分钟5万
};

export default function () {
    const url = '你的目标URL';
    const payload = JSON.stringify({ "固定payload": "xxx" });
    const params = { headers: { 'Content-Type': 'application/json' } };
    http.post(url, payload, params);
    sleep(0);  // 无等待
}

启动命令：k6 run script.js

• 关闭所有非必要的日志/打印操作，减少IO开销。
• 并发数不要盲目调得过高，避免本地机器端口耗尽或目标服务被打垮。
• 如果单台机器性能不够，专业工具都支持分布式压测，多台机器协同发起请求。

内容的提问来源于stack exchange，提问作者x89