JMeter工作流负载测试设计：如何并行执行子API工作流

核心逻辑 #

先调用父API获取子API列表，再根据列表动态生成并行执行单元，每个单元绑定对应子API的串行工作流（Thread Group），实现所有子API工作流的同时启动。

基于JMeter的实现步骤 #

如果使用JMeter作为测试/工作流执行工具，可按以下方式实现：

1. 获取子API列表

   • 添加HTTP请求Sampler调用父API，通过JSON Extractor（或正则提取器）解析返回结果，将子API ID列表存储为逗号分隔的变量（如child_api_ids）。

2. 动态创建并行Thread Group

   • 添加JSR223 Sampler（选用Groovy语言），编写脚本遍历子API ID列表，为每个ID创建独立的Thread Group，并传递子API ID到对应线程组：

     def childIds = vars.get("child_api_ids")?.split(",") ?: []
     def testPlan = ctx.getThreadGroup().getTestPlan()
     def templateGroup = ctx.getThreadGroup()  // 预先定义好的子API工作流线程组模板
     
     childIds.each { id ->
         def threadGroup = templateGroup.clone()
         threadGroup.setName("Child API ${id} Execution Group")
         threadGroup.setNumThreads(1)
         threadGroup.setRampUp(0)  // 立即启动，保证并行性
         threadGroup.getVariables().put("target_child_id", id)
     
         testPlan.addTestElement(threadGroup)
     }
     

   注意：需预先创建一个包含子API工作流逻辑的Thread Group作为模板，脚本通过克隆模板来生成新线程组，避免重复编写工作流代码。

3. 资源适配（针对大量子API场景）

   • 当子API数量达1000个时，需调整JMeter堆内存（修改jmeter.bat/jmeter.sh中的HEAP参数，如-Xms2g -Xmx10g）；
   • 若直接启动1000个线程导致资源耗尽，可改为分批并行，通过设置线程组的启动延迟，或使用Throughput Controller控制并发数。

通用代码实现方案（自定义工作流） #

如果是自行开发工作流系统，可基于线程池实现并行执行：
以Python为例：

import concurrent.futures

def execute_child_workflow(child_api_id):
    # 这里是子API对应的串行工作流逻辑
    # 示例：调用子API、执行后续步骤
    print(f"启动子API {child_api_id} 的工作流")
    # 实际逻辑：requests.get(f"https://your-api.com/child/{child_api_id}")
    # ... 其他串行步骤

# 1. 调用父API获取子API列表
parent_api_response = # 此处实现父API调用逻辑，返回子API ID列表
child_api_ids = parent_api_response.get("child_ids", [])

# 2. 用线程池并行执行工作流
# 线程数根据服务器资源调整，1000个任务建议设置100-200的线程数
with concurrent.futures.ThreadPoolExecutor(max_workers=150) as executor:
    # 提交所有子API工作流任务
    task_futures = [executor.submit(execute_child_workflow, api_id) for api_id in child_api_ids]
    
    # 等待所有任务完成，处理结果或异常
    for future in concurrent.futures.as_completed(task_futures):
        try:
            result = future.result()
            # 处理工作流执行结果
        except Exception as e:
            print(f"子API工作流执行失败: {str(e)}")

关键：通过线程池控制并发数，平衡并行效率与资源占用，避免因线程过多导致系统崩溃。

内容的提问来源于stack exchange，提问作者nsrsairajesh