JMeter如何配置Throughput Shaping Timer实现TPS无限阶梯递增压测

你当前使用的JMeter 5.2.1版本搭配Concurrency Thread Group、Throughput Shaping Timer（以下简称TST）的组合可以实现该需求，具体配置步骤如下：

1. 基础组件保留配置 #

Concurrency Thread Group的「Target Concurrency」字段继续保留原参数：
${__tstFeedback(ThroughputShapingTimer,1,10,10)}

提示：可根据你预期的最大TPS调整函数的第三个参数（最大线程数），避免线程资源不足导致TPS无法达预期。

2. 全局变量定义 #

在测试计划根节点下添加「用户定义的变量」组件，配置3个可自定义的参数：

• 初始TPS_X：你需要的起始TPS值
• 单阶梯持续时间_Y分钟：每个TPS档位的保持时长
• 阶梯步长_Z：每次递增的TPS数值

3. 动态TPS规则实现 #

在线程组同级添加JSR223 PreProcessor，语言选择groovy，勾选「每次迭代更新」，填入以下代码：

import kg.apc.jmeter.timers.VariableThroughputTimer
import kg.apc.jmeter.timers.VariableThroughputTimer.ShapingPatternEntry

// 初始化全局阶梯计数
if (props.get("current_step") == null) {
    props.put("current_step", 0)
}

// 读取预设变量
def initTps = Integer.parseInt(vars.get("初始TPS_X"))
def stepDurationMin = Integer.parseInt(vars.get("单阶梯持续时间_Y分钟"))
def stepTps = Integer.parseInt(vars.get("阶梯步长_Z"))

// 计算当前阶梯目标TPS
def currentTargetTps = initTps + (props.get("current_step") as int) * stepTps
def stepDurationSec = stepDurationMin * 60

// 获取TST组件并更新负载规则
def tst = ctx.getTestPlan().findTestElementsOfType(VariableThroughputTimer.class, false).get(0)
def entries = []
entries.add(new ShapingPatternEntry(0, stepDurationSec, currentTargetTps))
tst.setData(entries)

// 阶梯执行完成后更新计数，下一轮自动升挡
if (ctx.getThread().getLastSampleEndTime() - ctx.getThread().getStartTime() >= stepDurationSec * 1000) {
    props.put("current_step", (props.get("current_step") as int) + 1)
}

4. 无限循环与停止条件配置 #

• 在Concurrency Thread Group外层套一个无限循环控制器，保证阶梯升挡逻辑可以重复执行
• 在压测请求下添加响应断言，按你的业务规则配置错误判定条件（如响应码非2xx、返回内容不匹配等）
• 在线程组同级添加JSR223 Listener，填入以下代码实现报错即停止测试：

if (!prev.isSuccessful()) {
    // 出现错误立即终止测试
    ctx.getEngine().stopTest()
}

负载模式效果参考 #

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