四、性能压测与容量规划

4.1 压力测试实践

# locustfile.py - 使用Locust进行压测
from locust import HttpUser, task, between, events
import random
import json

class OrderUser(HttpUser):
    """模拟用户行为"""
    wait_time = between(0.5, 2)  # 思考时间0.5-2秒

    def on_start(self):
        """用户启动时执行"""
        # 登录获取token
        response = self.client.post("/api/auth/login", json={
            "username": f"user_{random.randint(1, 10000)}",
            "password": "test123"
        })
        self.token = response.json().get("token")
        self.headers = {"Authorization": f"Bearer {self.token}"}

    @task(3)  # 权重3
    def view_products(self):
        """浏览商品"""
        self.client.get("/api/products", headers=self.headers)

    @task(2)
    def add_to_cart(self):
        """添加购物车"""
        product_id = random.randint(1, 1000)
        self.client.post(f"/api/cart/items", 
                        json={"product_id": product_id, "quantity": 1},
                        headers=self.headers)

    @task(1)
    def checkout(self):
        """下单"""
        self.client.post("/api/orders", 
                        json={"shipping_address": "Test Address"},
                        headers=self.headers)

# 运行压测
# locust -f locustfile.py --host=https://api.example.com --users=1000 --spawn-rate=10

4.2 容量规划公式

class CapacityPlanner:
    """容量规划器"""

    @staticmethod
    def calculate_qps_required(daily_requests, peak_factor=3):
        """
        计算所需QPS

        daily_requests: 每日总请求量
        peak_factor: 峰值系数（高峰是平峰的倍数）
        """
        # 假设80%的请求集中在20%的时间段
        peak_seconds = 24 * 3600 * 0.2  # 4.8小时
        peak_qps = (daily_requests * 0.8) / peak_seconds
        return peak_qps * peak_factor

    @staticmethod
    def calculate_servers_needed(peak_qps, single_server_qps, safety_factor=1.5):
        """计算所需服务器数量"""
        return math.ceil((peak_qps / single_server_qps) * safety_factor)

    @staticmethod
    def calculate_db_connections(qps, avg_query_time_ms, connection_pool_size):
        """计算数据库连接需求"""
        # 每个请求占用连接的时间 = 查询时间 * (1 + 等待因子)
        avg_hold_time = avg_query_time_ms / 1000 * 1.2
        # 所需连接数 = QPS * 平均持有时间
        required = qps * avg_hold_time
        return max(required, connection_pool_size)

    @staticmethod
    def calculate_cache_size(daily_active_users, avg_data_per_user_kb):
        """计算缓存容量需求（KB）"""
        # 缓存热点数据：20%的用户贡献80%的请求
        hot_users = daily_active_users * 0.2
        return hot_users * avg_data_per_user_kb

# 示例：规划双十一容量
daily_requests = 100_000_000  # 1亿请求
peak_qps = CapacityPlanner.calculate_qps_required(daily_requests, peak_factor=2)
print(f"峰值QPS: {peak_qps:.0f}")  # 约46,296 QPS

servers = CapacityPlanner.calculate_servers_needed(peak_qps, 1000, 1.5)
print(f"需要服务器: {servers} 台")  # 约70台

五、总结

5.1 核心能力图谱

5.2 优化优先级矩阵

                    影响范围
                小              大
         ┌─────────────┬─────────────┐
      高 │   立即处理   │   最高优先   │
         │  代码级优化  │  架构级优化  │
  复     ├─────────────┼─────────────┤
  杂     │   低优先级   │   计划处理   │
  度  低 │  配置调优   │  基础设施升级│
         └─────────────┴─────────────┘

记住：性能、稳定性、安全是系统生命线，缺一不可。优秀的工程师不仅能把功能做对，更能把系统做好。持续优化、持续改进，让每一行代码都更有价值。
来源：
https://lemci.cn/