如何编写接口压测脚本

wen 实用脚本 2026-06-30 134

本文目录导读：

如何编写接口压测脚本

使用 Python + Locust
使用 Python + aiohttp (异步高并发)
使用 Apache JMeter (GUI工具)
使用 k6 (高性能压测工具)
常见压测脚本模板
性能测试报告生成
最佳实践建议

我来详细介绍如何编写接口压测脚本,包含多种常见工具和方法。

使用 Python + Locust

安装

pip install locust

基础脚本示例

from locust import HttpUser, task, between
class WebsiteUser(HttpUser):
    wait_time = between(1, 2)  # 用户请求间隔
    @task
    def get_users(self):
        self.client.get("/api/users")
    @task(3)  # 权重为3，执行概率更高
    def create_user(self):
        payload = {
            "name": "test_user",
            "email": "test@example.com"
        }
        self.client.post("/api/users", json=payload)
    def on_start(self):
        """每个用户启动时执行"""
        self.client.post("/login", {
            "username": "test",
            "password": "123456"
        })

运行命令

# Web界面模式
locust -f locustfile.py --host=http://example.com
# 无头模式
locust -f locustfile.py --headless -u 100 -r 10 --run-time 1m

使用 Python + aiohttp (异步高并发)

import asyncio
import aiohttp
import time
import statistics
class PerformanceTester:
    def __init__(self, base_url, concurrency=50):
        self.base_url = base_url
        self.concurrency = concurrency
        self.results = []
    async def single_request(self, session, endpoint, method='GET', data=None):
        start_time = time.time()
        try:
            url = f"{self.base_url}{endpoint}"
            if method == 'GET':
                async with session.get(url) as response:
                    status = response.status
            else:
                async with session.post(url, json=data) as response:
                    status = response.status
            elapsed = time.time() - start_time
            self.results.append({
                'status': status,
                'time': elapsed,
                'success': status == 200
            })
            return elapsed
        except Exception as e:
            self.results.append({
                'status': 0,
                'time': time.time() - start_time,
                'success': False,
                'error': str(e)
            })
    async def run_test(self, endpoint, total_requests=1000):
        connector = aiohttp.TCPConnector(limit=self.concurrency)
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = []
            for _ in range(total_requests):
                task = self.single_request(session, endpoint)
                tasks.append(task)
            await asyncio.gather(*tasks)
    def report(self):
        successful = [r for r in self.results if r['success']]
        failed = [r for r in self.results if not r['success']]
        times = [r['time'] for r in successful]
        print(f"总请求数: {len(self.results)}")
        print(f"成功请求: {len(successful)}")
        print(f"失败请求: {len(failed)}")
        if times:
            print(f"平均响应时间: {statistics.mean(times):.3f}s")
            print(f"最大响应时间: {max(times):.3f}s")
            print(f"最小响应时间: {min(times):.3f}s")
            print(f"P50: {sorted(times)[len(times)//2]:.3f}s")
            print(f"P99: {sorted(times)[int(len(times)*0.99)]:.3f}s")
            print(f"QPS: {len(successful)/sum(times):.2f}")
# 使用示例
async def main():
    tester = PerformanceTester("http://api.example.com", concurrency=100)
    await tester.run_test("/api/endpoint", total_requests=5000)
    tester.report()
asyncio.run(main())

使用 Apache JMeter (GUI工具)

创建测试计划

<?xml version="1.0" encoding="UTF-8"?>
<jmeterTestPlan version="1.2">
  <hashTree>
    <TestPlan guiclass="TestPlanGui" testclass="TestPlan" testname="API压力测试">
      <elementProp name="TestPlan.user_defined_variables" elementType="Arguments">
        <collectionProp name="Arguments.arguments">
          <elementProp name="base_url" elementType="Argument">
            <stringProp name="Argument.name">base_url</stringProp>
            <stringProp name="Argument.value">http://api.example.com</stringProp>
          </elementProp>
        </collectionProp>
      </elementProp>
    </TestPlan>
    <hashTree>
      <ThreadGroup guiclass="ThreadGroupGui" testclass="ThreadGroup" testname="用户组">
        <stringProp name="ThreadGroup.num_threads">100</stringProp>
        <stringProp name="ThreadGroup.ramp_time">10</stringProp>
        <stringProp name="ThreadGroup.loop">-1</stringProp>
      </ThreadGroup>
      <hashTree>
        <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API请求">
          <stringProp name="HTTPSampler.domain">${base_url}</stringProp>
          <stringProp name="HTTPSampler.path">/api/endpoint</stringProp>
          <stringProp name="HTTPSampler.method">GET</stringProp>
        </HTTPSamplerProxy>
        <hashTree/>
        <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="固定定时器">
          <stringProp name="ConstantTimer.delay">1000</stringProp>
        </ConstantTimer>
        <hashTree/>
      </hashTree>
    </hashTree>
  </hashTree>
</jmeterTestPlan>

JMeter CLI 运行

jmeter -n -t test_plan.jmx -l results.jtl

使用 k6 (高性能压测工具)

// stress_test.js
import http from 'k6/http';
import { check, sleep } from 'k6';
import { Rate, Trend } from 'k6/metrics';
// 自定义指标
const errorRate = new Rate('errors');
const responseTime = new Trend('response_time');
export const options = {
  stages: [
    { duration: '2m', target: 100 }, // 逐渐增加到100用户
    { duration: '5m', target: 100 }, // 保持100用户
    { duration: '2m', target: 0 },   // 逐渐减少
  ],
  thresholds: {
    http_req_duration: ['p(95)<500'], // 95%的请求在500ms内
    errors: ['rate<0.1'],             // 错误率小于10%
  },
};
export default function () {
  const baseUrl = 'http://api.example.com';
  // GET请求
  const response = http.get(`${baseUrl}/api/users`);
  // 检查响应
  check(response, {
    '状态码是200': (r) => r.status === 200,
    '响应时间小于2s': (r) => r.timings.duration < 2000,
  });
  // 记录错误
  errorRate.add(response.status !== 200);
  responseTime.add(response.timings.duration);
  sleep(1);
}

运行k6

k6 run stress_test.js

常见压测脚本模板

带认证的压测脚本

# locust_with_auth.py
from locust import HttpUser, task, between
import random
class AuthenticatedUser(HttpUser):
    wait_time = between(0.5, 1.5)
    def on_start(self):
        # 获取token
        response = self.client.post("/api/login", {
            "username": "admin",
            "password": "password123"
        })
        self.token = response.json().get("token")
        self.headers = {"Authorization": f"Bearer {self.token}"}
    @task
    def search_products(self):
        params = {
            "q": random.choice(["phone", "laptop", "book"]),
            "page": random.randint(1, 10),
            "limit": 20
        }
        with self.client.get(
            "/api/products",
            params=params,
            headers=self.headers,
            catch_response=True
        ) as response:
            if response.status_code != 200:
                response.failure(f"Got status {response.status_code}")
            elif response.elapsed.total_seconds() > 5:
                response.failure("Response too slow")

混合请求类型

from locust import HttpUser, task, between
import json
class MixedLoadUser(HttpUser):
    wait_time = between(0.5, 3)
    @task(5)  # 高频率
    def view_products(self):
        self.client.get("/api/products/1")
    @task(3)  # 中等频率
    def add_to_cart(self):
        self.client.post("/api/cart/add", json={
            "product_id": 123,
            "quantity": 1
        })
    @task(1)  # 低频率
    def checkout(self):
        self.client.post("/api/orders", json={
            "items": [{"product_id": 123, "quantity": 1}],
            "payment_method": "credit_card"
        })

性能测试报告生成

生成HTML报告 (Locust)

locust -f locustfile.py --host=http://example.com --csv=mytest --html=report.html

自定义报告函数

def generate_report(results, output_file='performance_report.txt'):
    with open(output_file, 'w') as f:
        f.write("=" * 50 + "\n")
        f.write("性能测试报告\n")
        f.write("=" * 50 + "\n\n")
        # 基本统计
        f.write(f"测试时间: {results['start_time']} - {results['end_time']}\n")
        f.write(f"总请求数: {results['total_requests']}\n")
        f.write(f"并发用户数: {results['concurrency']}\n\n")
        # 响应时间分布
        f.write("响应时间分布:\n")
        for percentile in [50, 75, 90, 95, 99]:
            time = results['percentiles'].get(percentile, 0)
            f.write(f"  P{percentile}: {time:.2f}ms\n")
        f.write(f"\n最大响应时间: {results['max_time']:.2f}ms\n")
        f.write(f"最小响应时间: {results['min_time']:.2f}ms\n")
        f.write(f"平均响应时间: {results['avg_time']:.2f}ms\n")
        f.write(f"吞吐量(QPS): {results['qps']:.2f}\n")
        # 错误统计
        f.write(f"\n错误统计:\n")
        f.write(f"  总错误数: {results['error_count']}\n")
        f.write(f"  错误率: {results['error_rate']:.2%}\n")
        if results['errors']:
            f.write("\n错误详情:\n")
            for error_type, count in results['errors'].items():
                f.write(f"  {error_type}: {count}\n")
    print(f"报告已生成: {output_file}")

最佳实践建议

压测前准备

设置合理的参数
- 从低并发开始,逐步增加
- 设置超时时间,避免请求挂起

数据准备

# 准备测试数据
test_data = {
 "users": [{"id": i, "name": f"user_{i}"} for i in range(100)],
 "tokens": ["token_" + str(i) for i in range(50)]
}

监控系统状态
```
import psutil
import time
```

def monitor_system(): """监控系统资源使用情况""" while True: cpu_percent = psutil.cpu_percent(interval=1) memory = psutil.virtual_memory() print(f"CPU: {cpu_percent}%, Memory: {memory.percent}%") time.sleep(5)


### 常见问题处理
- **连接池耗尽**: 增加连接池大小或减少并发数
- **内存泄漏**: 监控内存使用，添加内存限制
- **超时**: 设置合理的超时时间，区分网络超时和业务超时
选择适合你项目的工具和方法，建议从简单的脚本开始，逐步完善压测方案。

本文地址： https://www.bmjup.com/post/12243.html