25000人机交互测试入门指南：从零搭建高并发测试框架

1次阅读

共计 2597 个字符，预计需要花费 7 分钟才能阅读完成。

大规模人机交互测试（Human-Machine Interaction Testing）面临三个核心挑战：

资源调度：单机难以支撑 25000 并发用户（Concurrent Users），需要分布式集群协调
结果一致性：高并发下请求顺序紊乱可能导致断言失败（Assertion Failure）
性能监控：传统监控工具在超大规模测试时会产生额外性能开销（Overhead）

工具	并发控制能力	CPU 占用率（单节点）	报告可视化
JMeter	依赖线程模型	高（1 万并发约 75%）	需插件扩展
Locust	协程异步 IO	低（1 万并发约 30%）	内置实时图表
Gatling	Actor 模型	中（1 万并发约 50%）	专业 HTML 报告

选择 Locust 的核心优势：

Python 生态便于扩展
支持分布式压测（Distributed Load Testing）
零配置实现 Web 可视化控制

from locust import HttpUser, task, between
from typing import Optional
import redis

class BoundedRedis(redis.Redis):
    """带连接池管理的 Redis 客户端"""
    def __del__(self):
        self.connection_pool.disconnect()

class DemoUser(HttpUser):
    wait_time = between(1, 3)  # 思考时间(Wait Time)
    host = "http://api.example.com"

    def on_start(self):
        """模拟登录会话(Session)"""
        self.client.post("/login", json={"user": "test"})

    @task(weight=3)  # 权重设置
    def get_index(self):
        with self.client.get("/", catch_response=True) as resp:
            if resp.status_code != 200:
                resp.failure("Status error")

    @task(weight=1)
    def post_data(self):
        payload = {"key": self._get_redis_data()}  # 共享测试数据
        self.client.post("/submit", json=payload)

    def _get_redis_data(self) -> Optional[str]:
        """从 Redis 集群获取测试数据"""
        try:
            r = BoundedRedis(host='redis-master')
            return r.get("test_key").decode()
        except Exception as e:
            self.environment.events.request_failure.fire(request_type="Redis", name="cache_get", response_time=0, exception=e)
            return None

会话保持：
每个虚拟用户 (Virtual User) 独立 Cookie
使用 Redis 存储全局 Session Token
分布式架构：

flowchart TB
    master[Master 节点] -->| 协调 | worker1[Worker 1]
    master -->| 分配用户 | worker2[Worker 2]
    worker1 -->| 压测流量 | target[被测系统]
    worker2 -->| 压测流量 | target

启动命令：

# Master 节点
locust -f test.py --master --expect-workers 10

# Worker 节点（需启动 10 个）locust -f test.py --worker --master-host=<MASTER_IP>

# 使用 Hash Slot 分片存储
r = redis.RedisCluster(startup_nodes=[{"host": "redis-node1", "port": "6379"}],
    decode_responses=True
)

# 写入测试数据时采用轮询分布
for i in range(10000):
    r.set(f"test_{i%1000}", mock_data())  # 分片到不同 Slot

Prometheus 配置：

scrape_configs:
  - job_name: 'locust'
    static_configs:
      - targets: ['worker1:8080', 'worker2:8080']

Grafana 看板指标：
Requests/s（每秒请求数）
95th Percentile Latency（P95 延迟）
Worker CPU/Memory Usage（节点资源使用率）

使用 tracemalloc 监控 Python 对象增长：

import tracemalloc
tracemalloc.start()

# 测试代码...
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
print("[Top 10 内存占用]", *top_stats[:10], sep="\n")