load-testing

负载测试

k6（推荐）

// load-test.js
import http from 'k6/http';
import { check, sleep } from 'k6';

export const options = {
  stages: [
    { duration: '1m', target: 50 },   // Ramp up to 50 users
    { duration: '3m', target: 50 },   // Stay at 50
    { duration: '1m', target: 100 },  // Ramp to 100
    { duration: '3m', target: 100 },  // Stay at 100
    { duration: '1m', target: 0 },    // Ramp down
  ],
  thresholds: {
    http_req_duration: ['p(95)<500', 'p(99)<1000'], // 95th < 500ms
    http_req_failed: ['rate<0.01'],                  // <1% error rate
  },
};

export default function () {
  const res = http.get('https://api.example.com/products');
  check(res, {
    'status is 200': (r) => r.status === 200,
    'response time < 500ms': (r) => r.timings.duration < 500,
  });
  sleep(1);
}

经过身份验证的请求

import http from 'k6/http';

export function setup() {
  const res = http.post('https://api.example.com/auth/login', JSON.stringify({
    email: '[email protected]', password: 'test-password',
  }), { headers: { 'Content-Type': 'application/json' } });
  return { token: res.json('token') };
}

export default function (data) {
  const headers = { Authorization: `Bearer ${data.token}`, 'Content-Type': 'application/json' };

  // Simulate user journey
  http.get('https://api.example.com/products', { headers });
  sleep(2);
  http.get('https://api.example.com/products/1', { headers });
  sleep(1);
  http.post('https://api.example.com/orders', JSON.stringify({ productId: '1', qty: 2 }), { headers });
}

Artillery

# artillery.yml
config:
  target: "https://api.example.com"
  phases:
    - duration: 60
      arrivalRate: 10
      name: "Warm up"
    - duration: 120
      arrivalRate: 50
      name: "Sustained load"
  defaults:
    headers:
      Content-Type: "application/json"

scenarios:
  - name: "Browse and purchase"
    flow:
      - get:
          url: "/products"
          capture:
            - json: "$.data[0].id"
              as: "productId"
      - think: 2
      - get:
          url: "/products/{{ productId }}"
      - think: 1
      - post:
          url: "/orders"
          json:
            productId: "{{ productId }}"
            quantity: 1

Locust（Python）

from locust import HttpUser, task, between

class WebUser(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        res = self.client.post("/auth/login", json={
            "email": "[email protected]", "password": "password"
        })
        self.token = res.json()["token"]
        self.headers = {"Authorization": f"Bearer {self.token}"}

    @task(3)
    def browse_products(self):
        self.client.get("/products", headers=self.headers)

    @task(1)
    def create_order(self):
        self.client.post("/orders", json={"productId": "1", "qty": 1}, headers=self.headers)

测试类型

类型	用途	时长
冒烟测试	验证系统在最低负载下是否正常运行	1-2 分钟
负载测试	预期的正常/峰值流量	10-30 分钟
压力测试	超出正常容量，找出系统崩溃点	10-20 分钟
浸泡测试	持续施加负载以查找内存泄漏	1-4 小时
峰值测试	突发流量	5-10 分钟

反模式

反模式	修复方法
从与服务相同的网络进行测试	从外部网络或云端进行测试
请求之间没有思考时间	添加符合实际情况的 sleep/think 延迟
只测试单个端点	测试符合实际情况的用户旅程
未定义阈值	设置 p95/p99 延迟和错误率阈值
对生产环境运行负载测试	使用预发布环境

生产检查清单

• 符合实际情况的用户场景（而不只是单个端点）
• 延迟（p95、p99）和错误率阈值
• 与预期流量形态相匹配的渐增模式
• 准备好测试数据（不重复使用同一条记录）
• 集成 CI/CD 以检测回归
• 归档结果，以便随时间进行比较