JMeter动态实现多用户按指定频率批量POST消息方案咨询

一、核心优化方向 #

1. 用JMeter全局变量替代硬编码，实现所有参数动态配置
2. 基于Constant Throughput Timer精准控制整体吞吐量，结合线程内间隔实现精细化调度
3. 重构文件分配逻辑，确保每个线程处理唯一文件组，彻底避免重复提交

二、具体配置步骤 #

1. 定义全局配置变量 #

添加用户定义的变量组件，统一管理所有可配置参数：

• total_messages：总消息发送量（如5000）
• threads：并发线程数（如20）
• throughput：每秒消息发送频率（如20）
• thread_interval：线程内消息发送间隔（毫秒，如5000）
• token_api_url：Token接口地址
• post_api_url：POST接口地址
• file_directory：本地文本文件目录

2. 文件唯一分配逻辑优化 #

原代码的threadNum % fileList.size()会导致文件重复读取，改为按线程分批分配文件：

// 替换原文件分配逻辑
int fileCount = fileList.size();
int filesPerThread = (int) Math.ceil((double) totalMessages / threads);
int startIndex = threadNum * filesPerThread;

// 循环处理当前线程专属的文件组
for (int i = startIndex; i < startIndex + filesPerThread && i < totalMessages && i < fileCount; i++) {
    File file = fileList.get(i);
    // 执行POST请求逻辑
}

同时修复文件排序逻辑，避免10a.txt排在2a.txt之前：

// 按文件名中的数字排序
fileList.sort((f1, f2) -> {
    Integer num1 = Integer.parseInt(f1.getName().replaceAll("[^0-9]", ""));
    Integer num2 = Integer.parseInt(f2.getName().replaceAll("[^0-9]", ""));
    return num1.compareTo(num2);
});

3. Constant Throughput Timer吞吐量控制 #

• 在线程组下添加Constant Throughput Timer，选择Calculate Throughput based on为All active threads (shared)
• 吞吐量值设置为${__javaScript(${throughput}*60,)}（JMeter默认单位是每分钟请求数，需转换）
• 若需线程内自定义间隔，在JSR223采样器中添加Thread.sleep(${thread_interval})，注意不要与Timer的全局控制冲突

4. Token全局缓存优化 #

用Once Only Controller实现Token全局复用，避免每个线程重复请求：

• 在线程组最外层添加Once Only Controller
• 内部添加JSR223采样器，请求Token并存入全局属性：props.put("authToken", authToken)
• POST请求采样器中通过props.get("authToken")获取Token

5. 总消息数循环控制 #

添加Loop Controller到线程组下，循环次数设置为${__javaScript(Math.ceil(${total_messages}/${threads}),)}，确保线程循环处理分配的消息量，直到完成总任务。

三、优化后的完整JSR223采样器代码 #

import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.FileEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.HttpResponse;
import org.apache.http.util.EntityUtils;

import java.io.File;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;

// 读取全局配置变量
int totalMessages = Integer.parseInt(vars.get("total_messages"));
int threads = Integer.parseInt(vars.get("threads"));
long threadInterval = Long.parseLong(vars.get("thread_interval"));
String tokenApiUrl = vars.get("token_api_url");
String postApiUrl = vars.get("post_api_url");
String fileDirectory = vars.get("file_directory");

SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
CloseableHttpClient httpClient = HttpClients.createDefault();
StringBuilder threadLog = new StringBuilder();
int threadNum = ctx.getThreadNum();
String logPrefix = String.format("%s INFO Thread %d", sdf.format(new Date()), threadNum);

try {
    // 获取全局缓存的Token
    String authToken = props.get("authToken").toString();

    // 读取并排序文件
    File dir = new File(fileDirectory);
    File[] files = dir.listFiles((d, name) -> name.endsWith(".txt"));
    List<File> fileList = new ArrayList<>();
    Collections.addAll(fileList, files);

    // 按文件名数字排序，保证顺序正确
    fileList.sort((f1, f2) -> {
        Integer num1 = Integer.parseInt(f1.getName().replaceAll("[^0-9]", ""));
        Integer num2 = Integer.parseInt(f2.getName().replaceAll("[^0-9]", ""));
        return num1.compareTo(num2);
    });

    int fileCount = fileList.size();
    int filesPerThread = (int) Math.ceil((double) totalMessages / threads);
    int startIndex = threadNum * filesPerThread;

    // 处理当前线程分配的文件
    for (int i = startIndex; i < startIndex + filesPerThread && i < totalMessages && i < fileCount; i++) {
        File file = fileList.get(i);
        threadLog.append(String.format("%s 开始处理文件: %s%n", logPrefix, file.getName()));

        long requestStartTime = System.currentTimeMillis();
        HttpPost postRequest = new HttpPost(postApiUrl);
        postRequest.setEntity(new FileEntity(file, "text/plain"));
        postRequest.setHeader("Content-type", "text/plain");
        postRequest.setHeader("Authorization", "Bearer " + authToken);
        postRequest.setHeader("newcustomerheader", "XXX");

        int maxRetries = 3;
        boolean success = false;
        String responseString = "";
        for (int retry = 0; retry < maxRetries; retry++) {
            HttpResponse response = httpClient.execute(postRequest);
            int statusCode = response.getStatusLine().getStatusCode();
            responseString = EntityUtils.toString(response.getEntity()).trim();

            if (statusCode >= 200 && statusCode < 300) {
                success = true;
                break;
            }
        }

        long requestEndTime = System.currentTimeMillis();
        long requestDuration = requestEndTime - requestStartTime;

        if (success) {
            threadLog.append(String.format("%s 处理成功: %s，响应ID: %s，耗时: %dms%n",
                    logPrefix, file.getName(), responseString, requestDuration));
        } else {
            threadLog.append(String.format("%s 处理失败(%d次重试): %s，响应: %s%n",
                    logPrefix, maxRetries, file.getName(), responseString));
        }

        // 线程内自定义间隔
        if (threadInterval > 0) {
            Thread.sleep(threadInterval);
        }
    }
} catch (Exception e) {
    threadLog.append(String.format("%s 发生错误: %s%n", logPrefix, e.getMessage()));
} finally {
    if (httpClient != null) {
        httpClient.close();
    }
    synchronized (this) {
        log.info(threadLog.toString());
    }
}

四、需求对应配置示例 #

示例1：用户数=1 → 每秒1条 #

• threads=1，throughput=1，total_messages=目标条数，thread_interval=0
• Timer吞吐量值自动转换为60（1*60）

示例2：总消息数=5000，频率=20条/秒，用户数=20，线程内每5秒发送一次 #

• threads=20，throughput=20，total_messages=5000，thread_interval=5000
• Loop Controller循环次数：${__javaScript(Math.ceil(5000/20),)}=250

示例3：总消息数=5000，频率=30条/秒，用户数=30 #

• threads=30，throughput=30，total_messages=5000，thread_interval=0
• Timer吞吐量值自动转换为1800（30*60）

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