在 Python 中实现压力测试

在 Python 中实现压力测试（性能测试或负载测试）通常涉及模拟高并发请求，以测试系统在高负载下的表现（如响应时间、吞吐量、资源占用等）。以下是常用的工具和方法：

1. 常用工具

a. Locust

• 简介：基于 Python 的开源负载测试工具，支持分布式测试，可通过代码定义用户行为。
• 安装：

  pip install locust
  

• 示例脚本 (locustfile.py)：

  from locust import HttpUser, task, between
  
  class WebsiteUser(HttpUser):
      wait_time = between(1, 5)  # 用户请求间隔时间
  
      @task
      def load_homepage(self):
          self.client.get("/")  # 测试访问根路径
  
      @task(3)  # 权重为3，更频繁执行
      def load_api(self):
          self.client.get("/api/data")
  

• 运行：

  locust -f locustfile.py
  

  访问 http://localhost:8089 配置并发用户数和速率。

b. JMeter + Python

• Jmeter：Java 编写的强大负载测试工具，支持通过 JSR223 调用 Python 脚本。
• 使用场景：复杂场景测试（如动态参数、逻辑分支）。

c. requests + 多线程/多进程

• 手动实现简单压测：

  import requests
  import threading
  
  def send_request(url):
      try:
          response = requests.get(url)
          print(f"Status Code: {response.status_code}")
      except Exception as e:
          print(f"Error: {e}")
  
  if __name__ == "__main__":
      url = "http://example.com/api"
      threads = []
      for _ in range(100):  # 模拟100个并发
          thread = threading.Thread(target=send_request, args=(url,))
          threads.append(thread)
          thread.start()
  
      for thread in threads:
          thread.join()
  

d. aiohttp + asyncio（异步高性能）

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    url = "http://example.com/api"
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for _ in range(1000)]  # 1000个异步请求
        await asyncio.gather(*tasks)

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

2. 关键指标

• 响应时间：平均、P90/P95/P99 分位。
• 吞吐量（Throughput）：单位时间处理的请求数。
• 错误率：失败请求占比。
• 资源占用：CPU、内存、网络带宽等。

3. 注意事项

1. 逐步加压：从低并发开始，逐步增加，避免瞬间压垮系统。
2. 监控目标系统：使用工具（如 top, htop, nmon, Prometheus）监控服务器资源。
3. 绕过客户端限制：
   • 调整操作系统文件描述符限制：ulimit -n 100000。
   • 使用分布式测试（如 Locust 的 Worker 模式）。
4. 结果分析：使用工具生成的报告（如 Locust 的 Web 界面）或结合时序数据库（InfluxDB + Grafana）。

4. 优化建议

• 代码性能：避免阻塞操作，使用异步或协程。
• 网络优化：减少请求体积，启用压缩（如 gzip）。
• 服务端优化：缓存、数据库索引、负载均衡。

根据需求选择工具：简单场景可用 requests + 多线程，复杂场景推荐 Locust 或 JMeter。