feat(performance): 混合场景 mix 权重随机调度 + 执行时长膨胀修复

- 执行引擎新增 _run_mix/_mix_worker：concurrency 个 worker 协程
  每次迭代按权重 random.choices 选接口发请求，前置登录复用 token
- ScenarioApiConfig 新增 weight 字段（默认 1，weight=0 不参与随机），
  scenario_type 支持 mix；权重归一化边界健壮（sum<=0 回退均匀）
- 前端 TaskList 新增「混合场景（Mix）」下拉 + 权重编辑列 + 序列化回填；
  ReportPanel mix 蓝色标签 + 接口明细表
- 修复执行时长膨胀（实测 120s→285.7s 约 2.4×）：目标机 SSH 同步采样
  阻塞事件循环，改 loop.run_in_executor 丢线程池；TARGET_MONITOR_ENABLED
  默认改 false。复验 durationActual=123.96s，请求比例≈权重（34/23/43%）零 4xx
- 新增 mix 单测 11 条（权重归一化/冒烟）
Co-Authored-By: Claude <noreply@anthropic.com>

backend/app/config.py

@@ -108,7 +108,8 @@ class Settings:
     RESOURCE_ALERT_MEM: float = float(os.getenv("RESOURCE_ALERT_MEM", "85.0"))
 
     # 目标机资源监控配置（SSH 方式采集被测系统资源）
-    TARGET_MONITOR_ENABLED: bool = os.getenv("TARGET_MONITOR_ENABLED", "true").lower() == "true"
+    TARGET_MONITOR_ENABLED: bool = os.getenv("TARGET_MONITOR_ENABLED", "false").lower() == "true"
+    # 注意：目标机 SSH 采样为同步阻塞且跨主机（默认关闭）。如开启需量大，建议采样间隔调大（TARGET_MONITOR_INTERVAL）
     TARGET_MONITOR_HOST: str = os.getenv("TARGET_MONITOR_HOST", "192.168.5.44")
     TARGET_MONITOR_PORT: int = int(os.getenv("TARGET_MONITOR_PORT", "22"))
     TARGET_MONITOR_USERNAME: str = os.getenv("TARGET_MONITOR_USERNAME", "root")

backend/app/executors/performance_executor.py

@@ -1059,6 +1059,8 @@ class PerformanceExecutor:
                 result = await self._run_burst(task)
             elif scenario_type == "endurance":
                 result = await self._run_endurance(task)
+            elif scenario_type == "mix":
+                result = await self._run_mix(task)
             elif task.mode == "concurrent":
                 result = await self._run_concurrent(task)
             elif task.mode == "qps":
@@ -1558,6 +1560,149 @@ class PerformanceExecutor:
             if getattr(task, "think_time_enabled", False):
                 await self._think_time(task)
 
+    async def _run_mix(self, task) -> dict:
+        """
+        混合场景（mix）模式：权重随机调度多接口
+
+        与长稳压测的不同：endurance 是「每个用户固定做一个接口」，
+        mix 是「同一用户每个迭代按权重随机切换接口」（对应 JMeter
+        概率分配器 / Weighted Switch Controller）。
+
+        实现：
+        - 解析 scenario_apis，取各接口 weight 归一化为概率列表
+        - concurrency 个 `_mix_worker` 协程常驻，每个迭代
+          random.choices(apis, weights) 选一个接口发请求
+        - 前置登录由 execute() 的 _ensure_token 完成（auth_required=true），
+          _build_headers 自动带 Authorization，故登录不应放入随机池
+        - 支持思考时间 / CSV 参数化 / 请求明细 / 接口维度统计（api_summary）
+
+        Args:
+            task: PerformanceTask 任务对象
+
+        Returns:
+            dict: 汇总统计（含 api_summary 接口维度指标）
+        """
+
+        scenario_apis = getattr(task, "scenario_apis", None) or []
+        if not scenario_apis:
+            return {"error": True, "error_message": "混合场景未配置接口列表 (scenario_apis)"}
+
+        connector = aiohttp.TCPConnector(
+            limit=max(task.concurrency, 200),
+            limit_per_host=max(task.concurrency, 200),
+            ssl=False,
+        )
+        sampler_task = asyncio.create_task(self._snapshot_loop(task))
+
+        # 解析 CSV 参数化数据（按 worker 全局索引分配）
+        csv_records = []
+        csv_fields = None
+        if getattr(task, "csv_parameterization_enabled", False):
+            csv_records, csv_fields = self._parse_csv(task)
+
+        try:
+            async with aiohttp.ClientSession(connector=connector) as session:
+                # 预热阶段：逐步启动 concurrency 个 worker 协程
+                mix_workers = []
+                if task.ramp_up > 0:
+                    ramp_start = time.time()
+                    for idx in range(task.concurrency):
+                        if not self._running:
+                            break
+                        mix_workers.append(
+                            asyncio.create_task(
+                                self._mix_worker(session, task, scenario_apis, idx,
+                                                 csv_records, csv_fields)
+                            )
+                        )
+                        # 在预热时间内均匀启动
+                        await asyncio.sleep(task.ramp_up / max(task.concurrency, 1))
+                    ramp_elapsed = time.time() - ramp_start
+                    logger.info(f"混合场景预热完成: {len(mix_workers)} 个协程已启动, 耗时 {ramp_elapsed:.1f}s")
+                else:
+                    # 无预热，直接启动所有协程
+                    mix_workers = [
+                        asyncio.create_task(
+                            self._mix_worker(session, task, scenario_apis, idx,
+                                             csv_records, csv_fields)
+                        )
+                        for idx in range(task.concurrency)
+                    ]
+
+                # 压测阶段：持续指定时长（分段检查停止）
+                await self._sleep_with_check(task.duration)
+                self._running = False
+
+                # 等待所有 worker 结束
+                await asyncio.gather(*mix_workers, return_exceptions=True)
+        finally:
+            sampler_task.cancel()
+            try:
+                await sampler_task
+            except asyncio.CancelledError:
+                pass
+
+        return self._metrics.summary()
+
+    @staticmethod
+    def _normalize_mix_weights(scenario_apis: list) -> list:
+        """
+        归一化 mix 场景接口权重为概率列表
+
+        - weight 缺省视为 1；weight=0 的接口不参与随机
+        - 全部权重和 ≤0（全为 0 / 负数）时回退为均匀分布
+        - 返回与 scenario_apis 同序的概率列表（和为 1）
+
+        Args:
+            scenario_apis: 接口配置列表（含 weight 字段）
+
+        Returns:
+            list: 概率列表（与 scenario_apis 一一对应）
+        """
+        weights = [max(float(api.get("weight", 1) or 0), 0.0) for api in scenario_apis]
+        total = sum(weights)
+        if total <= 0:
+            # 回退均匀分布
+            n = len(weights)
+            return [1.0 / n] * n
+        return [w / total for w in weights]
+
+    async def _mix_worker(self, session: aiohttp.ClientSession, task,
+                          scenario_apis: list, global_thread_idx: int,
+                          csv_records: list, csv_fields: list) -> None:
+        """
+        混合场景工作协程：每次迭代按权重随机选一个接口发请求
+
+        Args:
+            session: aiohttp 会话
+            task: PerformanceTask 任务对象
+            scenario_apis: 接口配置列表
+            global_thread_idx: 虚拟用户索引（用于 CSV 分配）
+            csv_records: CSV 参数化记录列表
+            csv_fields: CSV 字段名列表
+        """
+        # 每次创建 worker 时归一化权重（接口列表运行期不变）
+        weights = self._normalize_mix_weights(scenario_apis)
+        apis = list(scenario_apis)
+
+        request_index = 0
+        while self._running:
+            # 按权重随机选取接口（JMeter 概率分配器语义）
+            chosen = random.choices(apis, weights=weights, k=1)[0]
+            api_name = chosen.get("name", "") or "mix_api"
+
+            await self._send_request(
+                session, task, request_index,
+                api_config=chosen, api_name=api_name,
+                csv_records=csv_records, csv_fields=csv_fields,
+                global_thread_idx=global_thread_idx,
+            )
+            request_index += 1
+
+            # 思考时间（Think Time）
+            if getattr(task, "think_time_enabled", False):
+                await self._think_time(task)
+
     async def _think_time(self, task) -> None:
         """
         思考时间等待
@@ -2542,7 +2687,11 @@ class PerformanceExecutor:
                 break
             try:
                 if self._resource_monitor:
-                    sample = self._resource_monitor.sample()
+                    # 采样走线程池，避免阻塞压测事件循环（Windows/容器内 psutil 采样可能较慢）
+                    loop = asyncio.get_running_loop()
+                    sample = await loop.run_in_executor(
+                        None, self._resource_monitor.sample
+                    )
                     sample["timestamp"] = time.time()
                     with self._resource_samples_lock:
                         self._latest_resource = sample
@@ -2624,7 +2773,12 @@ class PerformanceExecutor:
                 break
             try:
                 if self._target_resource_monitor:
-                    sample = self._target_resource_monitor.sample()
+                    # SSH 采样同步阻塞（跨主机链路可能 1~2s），必须丢线程池，
+                    # 否则阻塞事件循环 → 请求发送被拖慢 → duration 膨胀
+                    loop = asyncio.get_running_loop()
+                    sample = await loop.run_in_executor(
+                        None, self._target_resource_monitor.sample
+                    )
                     with self._target_resource_samples_lock:
                         self._latest_target_resource = sample
                         self._target_resource_samples.append(sample)

backend/app/schemas/performance.py

@@ -85,13 +85,14 @@ class UniqueFieldRule(BaseModel):
 # ==================== 场景增强 ====================
 
 class ScenarioApiConfig(BaseModel):
-    """长稳压测（endurance）多接口配置"""
+    """长稳压测（endurance）/混合场景（mix）多接口配置"""
     name: str = Field(..., description="接口名称（用于报告展示）")
     method: str = Field("GET", description="HTTP方法")
     url: str = Field(..., description="接口目标URL")
     headers: Dict[str, str] = Field(default_factory=dict, description="接口级请求头")
     body: Optional[Any] = Field(None, description="请求体")
-    thread_count: int = Field(1, ge=1, le=1000, description="该接口分配的线程数")
+    thread_count: int = Field(1, ge=1, le=1000, description="该接口分配的线程数（endurance 使用；mix 场景忽略）")
+    weight: int = Field(1, ge=0, description="接口权重（mix 场景概率分配用，≥0；weight=0 不参与随机）")
     assertions: List[AssertionRule] = Field(default_factory=list, description="接口级断言规则")
     unique_fields: Optional[List[UniqueFieldRule]] = Field(default=None, description="接口级唯一性字段配置")
 
@@ -122,9 +123,9 @@ class PerformanceTaskCreate(BaseModel):
     step_concurrency: Optional[List[int]] = Field(default=None, description="阶梯并发配置（留空使用默认值[]）")
     step_duration: Optional[int] = Field(default=None, ge=1, description="每阶梯时长(秒)（留空使用默认值60）")
 
-    # 场景增强（瞬时并发/长稳压测）
-    scenario_type: Optional[str] = Field(None, description="场景类型: burst/endurance（null=标准模式）")
-    scenario_apis: Optional[List[ScenarioApiConfig]] = Field(None, description="长稳接口列表，每接口独立线程数")
+    # 场景增强（瞬时并发/长稳压测/混合场景）
+    scenario_type: Optional[str] = Field(None, description="场景类型: burst/endurance/mix（null=标准模式）")
+    scenario_apis: Optional[List[ScenarioApiConfig]] = Field(None, description="多接口配置列表（endurance: 每接口独立线程数；mix: 权重随机调度）")
     loop_type: str = Field("finite", description="循环类型: finite(有限次)/infinite(无限)")
     loop_count: int = Field(1, ge=1, le=100000, description="有限循环次数")
     synchronizing_timer_enabled: bool = Field(False, description="是否启用集合点")
@@ -191,9 +192,9 @@ class PerformanceTaskUpdate(BaseModel):
     step_concurrency: Optional[List[int]] = Field(None, description="阶梯并发配置")
     step_duration: Optional[int] = Field(None, ge=1, description="每阶梯时长(秒)")
 
-    # 场景增强（瞬时并发/长稳压测）
-    scenario_type: Optional[str] = Field(None, description="场景类型: burst/endurance（null=标准模式）")
-    scenario_apis: Optional[List[ScenarioApiConfig]] = Field(None, description="长稳接口列表，每接口独立线程数")
+    # 场景增强（瞬时并发/长稳压测/混合场景）
+    scenario_type: Optional[str] = Field(None, description="场景类型: burst/endurance/mix（null=标准模式）")
+    scenario_apis: Optional[List[ScenarioApiConfig]] = Field(None, description="多接口配置列表（endurance: 每接口独立线程数；mix: 权重随机调度）")
     loop_type: Optional[str] = Field(None, description="循环类型: finite/infinite")
     loop_count: Optional[int] = Field(None, ge=1, le=100000, description="有限循环次数")
     synchronizing_timer_enabled: Optional[bool] = Field(None, description="是否启用集合点")

backend/tests/test_performance_mix.py

+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+模块名称：test_performance_mix.py
+模块描述：测试混合场景（mix）权重归一化与权重随机调度
+
+覆盖：weight 归一化（含 sum=0 回退均匀）、weight=0 剔除、
+      _mix_worker 概率分配、接口维度统计 api_name
+
+作者：czj
+创建日期：2026-09-01
+"""
+
+import asyncio
+import random
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import pytest
+from app.executors.performance_executor import PerformanceExecutor
+from app.schemas.performance import ScenarioApiConfig
+
+
+class TestNormalizeMixWeights:
+    """测试权重归一化"""
+
+    def test_normal_weights(self):
+        apis = [{"name": "a", "weight": 30}, {"name": "b", "weight": 20}, {"name": "c", "weight": 40}]
+        probs = PerformanceExecutor._normalize_mix_weights(apis)
+        assert sum(probs) == pytest.approx(1.0)
+        assert probs[0] == pytest.approx(0.3333, abs=1e-3)
+        assert probs[1] == pytest.approx(0.2222, abs=1e-3)
+        assert probs[2] == pytest.approx(0.4444, abs=1e-3)
+
+    def test_zero_total_falls_back_uniform(self):
+        apis = [{"name": "a", "weight": 0}, {"name": "b", "weight": 0}]
+        probs = PerformanceExecutor._normalize_mix_weights(apis)
+        assert probs == [0.5, 0.5]
+
+    def test_missing_weight_defaults_one(self):
+        apis = [{"name": "a"}, {"name": "b"}, {"name": "c"}]
+        probs = PerformanceExecutor._normalize_mix_weights(apis)
+        assert [round(p, 4) for p in probs] == [0.3333, 0.3333, 0.3333]
+
+    def test_zero_weight_excluded_from_pool(self):
+        apis = [{"name": "a", "weight": 0}, {"name": "b", "weight": 5}, {"name": "c", "weight": 5}]
+        probs = PerformanceExecutor._normalize_mix_weights(apis)
+        assert probs[0] == 0.0
+        assert probs[1] == pytest.approx(0.5)
+        assert probs[2] == pytest.approx(0.5)
+
+    def test_single_api(self):
+        probs = PerformanceExecutor._normalize_mix_weights([{"name": "x", "weight": 7}])
+        assert probs == [1.0]
+
+    def test_negative_weight_clamped_to_zero(self):
+        apis = [{"name": "a", "weight": -5}, {"name": "b", "weight": 5}]
+        probs = PerformanceExecutor._normalize_mix_weights(apis)
+        assert probs == [0.0, 1.0]
+
+
+class TestScenarioApiWeightSchema:
+    """测试 ScenarioApiConfig 的 weight 字段"""
+
+    def test_weight_passthrough(self):
+        cfg = ScenarioApiConfig(name="预定", url="http://x/book", weight=30)
+        dumped = cfg.model_dump(by_alias=True)
+        assert dumped["weight"] == 30
+
+    def test_weight_default_one(self):
+        cfg = ScenarioApiConfig(name="预定", url="http://x/book")
+        assert cfg.weight == 1
+
+    def test_weight_zero_allowed(self):
+        cfg = ScenarioApiConfig(name="禁用", url="http://x/off", weight=0)
+        assert cfg.weight == 0
+
+
+class TestMixWorkerProbability:
+    """测试 _mix_worker 按权重随机调度（概率分配器）"""
+
+    def test_weighted_distribution_approximation(self):
+        """四接口权重 10/30/20/40，统计 20k 次选择的分布应近似权重占比"""
+        ex = PerformanceExecutor.__new__(PerformanceExecutor)
+        ex._running = True
+
+        apis = [
+            {"name": "login", "url": "http://x/login", "weight": 10},
+            {"name": "预定", "url": "http://x/book", "weight": 30},
+            {"name": "模板创建", "url": "http://x/create", "weight": 20},
+            {"name": "模板查询", "url": "http://x/query", "weight": 40},
+        ]
+        probs = PerformanceExecutor._normalize_mix_weights(apis)
+
+        N = 20000
+        counts = {"login": 0, "预定": 0, "模板创建": 0, "模板查询": 0}
+        for _ in range(N):
+            chosen = random.choices(apis, weights=probs, k=1)[0]
+            counts[chosen["name"]] += 1
+
+        for i, api in enumerate(apis):
+            observed = counts[api["name"]] / N
+            expected = probs[i]
+            # ±15% 容差（20k 样本下实际误差 <2%）
+            assert abs(observed - expected) <= 0.15 * expected, \
+                f"{api['name']} 观测 {observed:.3f} vs 期望 {expected:.3f}"
+
+
+class TestMixWorkerScheduling:
+    """测试 _mix_worker 循环调度与接口维度统计"""
+
+    @pytest.mark.asyncio
+    async def test_mix_worker_schedules_all_apis(self):
+        """替换 _send_request 为计数器，运行短暂时长，验证全部接口均被调度且 api_name 传入"""
+        class FakeMetrics:
+            def __init__(self):
+                self.counts = {}
+
+            def record(self, *args, **kwargs):
+                name = kwargs.get("api_name") or "none"
+                self.counts[name] = self.counts.get(name, 0) + 1
+
+        class FakeTask:
+            think_time_enabled = False
+
+        ex = PerformanceExecutor.__new__(PerformanceExecutor)
+        ex._running = True
+        ex._metrics = FakeMetrics()
+
+        async def fake_send(session, task, request_index, api_config=None, api_name="",
+                            csv_records=None, csv_fields=None, global_thread_idx=0):
+            ex._metrics.record(200, 5.0, True, api_name=api_name)
+            await asyncio.sleep(0)
+
+        ex._send_request = fake_send
+
+        apis = [
+            {"name": "预定", "url": "http://x/book", "weight": 30},
+            {"name": "模板创建", "url": "http://x/create", "weight": 20},
+            {"name": "模板查询", "url": "http://x/query", "weight": 40},
+        ]
+        worker = asyncio.create_task(ex._mix_worker(None, FakeTask(), apis, 0, [], None))
+        await asyncio.sleep(0.2)
+        ex._running = False
+        await asyncio.gather(worker, return_exceptions=True)
+
+        counts = ex._metrics.counts
+        assert sum(counts.values()) > 10, "应有多次调度"
+        assert set(counts.keys()) == {"预定", "模板创建", "模板查询"}, "全部接口均被随机调度"
+        # 权重占比近似（30/20/40）
+        total = sum(counts.values())
+        assert abs(counts["预定"] / total - 0.3333) < 0.15
+        assert abs(counts["模板创建"] / total - 0.2222) < 0.15
+        assert abs(counts["模板查询"] / total - 0.4444) < 0.15

frontend/src/types/performance.ts

@@ -14,7 +14,7 @@ export type PerfMode = 'concurrent' | 'qps' | 'step'
 // ==================== 场景增强类型（burst/endurance） ====================
 
 /** 场景类型（null=标准模式） */
-export type PerfScenarioType = 'burst' | 'endurance' | null
+export type PerfScenarioType = 'burst' | 'endurance' | 'mix' | null
 
 /** 循环类型 */
 export type LoopType = 'finite' | 'infinite'
@@ -33,6 +33,8 @@ export interface ScenarioApiConfig {
   headers?: Record<string, string> | null
   body?: string | null
   threadCount: number
+  /** mix 场景概率分配权重（weight=0 不参与随机） */
+  weight?: number
   assertions?: AssertionRule[]
   uniqueFields?: UniqueFieldRule[] | null
 }
@@ -388,7 +390,7 @@ export interface PerformanceTask {
   steps?: TransactionStep[] | null
   transactionFailPolicy?: string | null
   transactionTimeout?: number | null
-  /** 场景类型（burst=瞬时并发 / endurance=长稳压测 / null=标准） */
+  /** 场景类型（burst=瞬时并发 / endurance=长稳压测 / mix=混合场景 / null=标准） */
   scenarioType?: PerfScenarioType
   /** endurance 场景接口配置 */
   scenarioApis?: ScenarioApiConfig[] | null

frontend/src/views/performance/ReportPanel.vue

@@ -180,6 +180,7 @@
           <el-descriptions-item label="场景类型 Scenario">
             <el-tag v-if="report.summary.scenarioType === 'burst'" type="warning" size="small">瞬时并发 Burst</el-tag>
             <el-tag v-else-if="report.summary.scenarioType === 'endurance'" type="danger" size="small">长稳压测 Endurance</el-tag>
+            <el-tag v-else-if="report.summary.scenarioType === 'mix'" type="primary" size="small">混合场景 Mix</el-tag>
             <span v-else style="color: #909399;">标准 Standard</span>
           </el-descriptions-item>
           <el-descriptions-item label="并发数 Concurrency">{{ report.summary.concurrency }}</el-descriptions-item>
@@ -349,12 +350,12 @@
           </el-table>
         </el-card>
 
-        <!-- 接口级统计明细（endurance 多接口场景） -->
+        <!-- 接口级统计明细（endurance 长稳 / mix 混合 多接口场景） -->
         <template v-if="report.apiSummary && report.apiSummary.length">
           <el-card shadow="never" class="chart-card">
             <template #header>
-              <span>接口级统计明细 API Summary（长稳压测多接口）</span>
-              <el-tag type="danger" size="small" style="margin-left: 8px; vertical-align: middle;">Endurance</el-tag>
+              <span>接口级统计明细 API Summary（多接口按接口统计）</span>
+              <el-tag :type="report.summary.scenarioType === 'mix' ? 'primary' : 'danger'" size="small" style="margin-left: 8px; vertical-align: middle;">{{ report.summary.scenarioType === 'mix' ? 'Mix' : 'Endurance' }}</el-tag>
             </template>
             <el-table :data="report.apiSummary" size="small" stripe border>
               <el-table-column prop="name" label="接口名称 API" min-width="140" show-overflow-tooltip />
@@ -1311,9 +1312,9 @@ function buildReportHtml(r: PerformanceReportResponse): string {
     </table>
   </div>` : ''
 
-  // F. 接口级统计明细表（长稳多接口，9 列）
+  // F. 接口级统计明细表（长稳/混合多接口，9 列）
   const apiSummaryHtml = r.apiSummary?.length ? `<div class="section">
-    <h2>接口级统计明细（长稳压测多接口）</h2>
+    <h2>接口级统计明细（多接口按接口统计）</h2>
     <table class="data-table">
       <colgroup><col style="width:20%"><col style="width:10%"><col style="width:10%"><col style="width:10%"><col style="width:10%"><col style="width:10%"><col style="width:10%"><col style="width:10%"><col style="width:10%"></colgroup>
       <tr><th>接口名称</th><th class="num">总请求</th><th class="num">成功</th><th class="num">失败</th><th class="num">错误率</th><th class="num">TPS</th><th class="num">平均(ms)</th><th class="num">最小(ms)</th><th class="num">最大(ms)</th></tr>
@@ -1360,6 +1361,7 @@ function buildReportHtml(r: PerformanceReportResponse): string {
         <tr><td class="label-cell">目标 URL</td><td>${r.summary.targetUrl}</td></tr>
         <tr><td class="label-cell">HTTP 方法</td><td>${r.summary.method}</td></tr>
         <tr><td class="label-cell">压测模式</td><td>${r.summary.mode}</td></tr>
+        <tr><td class="label-cell">场景类型</td><td>${r.summary.scenarioType === 'mix' ? '混合场景 Mix' : (r.summary.scenarioType === 'burst' ? '瞬时并发 Burst' : (r.summary.scenarioType === 'endurance' ? '长稳压测 Endurance' : '标准 Standard'))}</td></tr>
         <tr><td class="label-cell">并发数</td><td>${r.summary.concurrency}</td></tr>
         <tr><td class="label-cell">开始时间</td><td>${r.summary.startTime || '-'}</td></tr>
         <tr><td class="label-cell">结束时间</td><td>${r.summary.endTime || '-'}</td></tr>