JMeter自定义Listener未将最后一批数据发送至InfluxDB问题

我使用带有自定义代码的JMeter Listener收集遥测数据，并将其批量推送至时序数据库InfluxDB以降低性能开销。当前脚本在批量条件满足时可正常推送数据，但测试执行结束时的最后一批数据无法推送至数据库。尝试使用if-else if-else逻辑处理该情况，但未生效。相关代码及执行日志如下：

自定义Listener代码 #

import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.params.BasicHttpParams;
import org.apache.http.util.EntityUtils;
import org.apache.jmeter.samplers.SampleResult;

try {

    Metrics = new StringBuilder();
    String status = "Failure";
    if (sampleResult.isSuccessful())
    status = "Success";
    if (sampleResult.getSampleLabel().startsWith("/"))
    Metrics.append("Requests")
    else
    Metrics.append("JMeter_Result")
    Metrics.append(",requestName=")
    .append(escapeValue(sampleResult.getSampleLabel()))
    .append(",httpstatuscode=")
    .append(sampleResult.getResponseCode())
    .append(",status=")
    .append(status)
    .append(",ScenarioID=")
    .append('${__groovy(vars.get("uniqueScenario_Id"),)}')
    .append(" responseTime=")
    .append(sampleResult.getTime())
    .append(",responsecode=")
    .append("\"${(sampleResult.getResponseCode())}\"")
    .append(",sentBytes=")
    .append(sampleResult.getSentBytes())
    .append(",receivedBytes=")
    .append(sampleResult.getBytesAsLong())
    .append(",responseMessage=")
    .append("\"${(sampleResult.getResponseMessage())}\"")
    .append(",samplecount=")
    .append(sampleResult.getSampleCount())
    .append(",errorCount=")
    .append(sampleResult.getErrorCount())
    .append(" ")
    .append(sampleResult.getTimeStamp())
    .append("000000");
    def measurementBody =  (ArrayList<String>) props.get("measurementBody");
    if(measurementBody == null)
            measurementBody = []
    measurementBody.add(Metrics);
    def NBody = [];
    if(measurementBody.size() > 6){
            log.info('Current Value :'+measurementBody.size()+' Condition Satisfied Data Pushed to influxDB ');
            props.put("measurementBody",NBody);
            PostMeasurement(measurementBody);
        }
        else if (sampleResult.getSampleLabel().equals("UserIdleTime")) {
            log.info('[Final]: As the idle situation arrived finall data pushed to influxDB');
            PostMeasurement(measurementBody);
        }
        
        else {
            log.info('Current Value :'+measurementBody.size()+' Condition Yet Satisfied Data Not Pushed to influxDB ');
            props.put("measurementBody",measurementBody)
        }
} catch (Exception e) {
    log.error("Error in Grafana Metrics Listener : " + e + "\n "+ e.getStackTrace());
}
//Escape the string values before posting the data
String escapeValue(String val) {
    String responseValueForSpaceRemoval;
    responseValueForSpaceRemoval = removeWhiteSpaces(val);
    val = responseValueForSpaceRemoval.replaceAll(",", "\\,")
    .replaceAll(" ", "\\")
    .replaceAll("=", "\\=")
    .replaceAll("\n", " ")
    .replaceAll("\r", " ")
    .replaceAll("\t", " ")
    .trim();
    return val;
}

//Handled all the sampler which have space between, this function will remove the space between
//[Note]:Space need to be remove as we are posting the data to influxdb & if your result have a space between then it will treat it as column value 
static String removeWhiteSpaces(String input) {
    return input.replaceAll(~/\s/, '')
}

// Method For Post the result to the time series database [influxDB]
public void PostMeasurement(ArrayList<String> Body) {
    def httpclient = new DefaultHttpClient(new BasicHttpParams());
    def httpPost = new HttpPost();
    httpPost.setURI(new URI(vars.get("InfluxDBAPI")));
    String stringBodyForDatabase = "";
    Body.each { stringBodyForDatabase = stringBodyForDatabase + it + "\n" };
    httpPost.setEntity(new StringEntity(stringBodyForDatabase))
    HttpResponse response = httpclient.execute(httpPost);
    EntityUtils.consumeQuietly(response.getEntity());
}

执行日志 #

2023-04-01 19:59:12,244 INFO o.a.j.e.StandardJMeterEngine: Running the test!
2023-04-01 19:59:12,244 INFO o.a.j.s.SampleEvent: List of sample_variables: []
2023-04-01 19:59:12,252 INFO o.a.j.g.u.JMeterMenuBar: setRunning(true, *local*)
2023-04-01 19:59:12,318 INFO o.a.j.e.StandardJMeterEngine: Starting ThreadGroup: 1 : 01_Study_Creation
2023-04-01 19:59:12,318 INFO o.a.j.e.StandardJMeterEngine: Starting 1 threads for group 01_Study_Creation.
2023-04-01 19:59:12,318 INFO o.a.j.e.StandardJMeterEngine: Thread will continue on error
2023-04-01 19:59:12,318 INFO o.a.j.t.ThreadGroup: Starting thread group... number=1 threads=1 ramp-up=1 delayedStart=false
2023-04-01 19:59:12,318 INFO o.a.j.t.ThreadGroup: Started thread group number 1
2023-04-01 19:59:12,318 INFO o.a.j.e.StandardJMeterEngine: All thread groups have been started
2023-04-01 19:59:12,320 INFO o.a.j.t.JMeterThread: Thread started: 01_Study_Creation 1-1
2023-04-01 19:59:14,842 INFO o.a.j.v.J.Request Sender to InfluxDB: Current Value :5 Condition Yet Satisfied Data Not Pushed to influxDB 
2023-04-01 19:59:21,485 INFO o.a.j.v.J.Request Sender to InfluxDB: Current Value :6 Condition Yet Satisfied Data Not Pushed to influxDB 
2023-04-01 19:59:22,508 INFO o.a.j.v.J.Request Sender to InfluxDB: Current Value :7 Condition Satisfied Data Pushed to influxDB 
2023-04-01 19:59:23,160 INFO o.a.j.v.J.Request Sender to InfluxDB: Current Value :1 Condition Yet Satisfied Data Not Pushed to influxDB 
2023-04-01 19:59:23,936 INFO o.a.j.v.J.Request Sender to InfluxDB: Current Value :2 Condition Yet Satisfied Data Not Pushed to influxDB 
2023-04-01 19:59:24,533 INFO o.a.j.v.J.Request Sender to InfluxDB: Current Value :3 Condition Yet Satisfied Data Not Pushed to influxDB 
2023-04-01 19:59:24,784 INFO o.a.j.v.J.Request Sender to InfluxDB: Current Value :4 Condition Yet Satisfied Data Not Pushed to influxDB 
2023-04-01 19:59:25,019 INFO o.a.j.v.J.Request Sender to InfluxDB: Current Value :5 Condition Yet Satisfied Data Not Pushed to influxDB 
2023-04-01 19:59:25,352 INFO o.a.j.v.J.Request Sender to InfluxDB: Current Value :6 Condition Yet Satisfied Data Not Pushed to influxDB 
2023-04-01 19:59:25,354 INFO o.a.j.t.JMeterThread: Thread is done: 01_Study_Creation 1-1
2023-04-01 19:59:25,354 INFO o.a.j.t.JMeterThread: Thread finished: 01_Study_Creation 1-1
2023-04-01 19:59:25,355 INFO o.a.j.e.StandardJMeterEngine: Notifying test listeners of end of test
2023-04-01 19:59:25,356 INFO o.a.j.g.u.JMeterMenuBar: setRunning(false, *local*)

问题分析 #

当前推送逻辑仅触发两种场景：

• 缓存数据量超过6条时（积累7条才推送）
• 采样器名称为UserIdleTime时

测试结束时，最后一批缓存数据量未达阈值，且没有UserIdleTime采样器触发，导致数据滞留在内存中无法推送。另外，代码未处理JMeter测试结束的生命周期事件。

解决方案 #

方案1：监听测试结束事件推送剩余数据 #

在自定义Listener中实现testEnded方法，测试结束时自动推送剩余数据。在现有代码末尾添加以下逻辑：

@Override
public void testEnded() {
    def measurementBody = (ArrayList<String>) props.get("measurementBody");
    if (measurementBody != null && !measurementBody.isEmpty()) {
        log.info('[Final]: Test ended, pushing remaining ' + measurementBody.size() + ' records to InfluxDB');
        PostMeasurement(measurementBody);
        props.put("measurementBody", []); // 清空缓存
    }
}

@Override
public void testEnded(String host) {
    testEnded(); // 统一处理本地和远程测试结束事件
}

方案2：添加收尾采样器触发推送 #

在测试计划最后添加一个Dummy Sampler，命名为UserIdleTime，确保测试执行到末尾时触发最后一批数据的推送逻辑。

额外优化建议 #

1. 替换过时的HTTP客户端：DefaultHttpClient已废弃，改用CloseableHttpClient避免资源泄漏：

import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;

// 修改PostMeasurement方法中的客户端初始化
def httpclient = HttpClients.createDefault();

2. 调整批量阈值逻辑：若想满6条就推送，将measurementBody.size() > 6改为measurementBody.size() >= 6
3. 优化字符串拼接：在PostMeasurement中用StringBuilder替代+=提升效率：

StringBuilder stringBodyForDatabase = new StringBuilder();
Body.each { stringBodyForDatabase.append(it).append("\n") };

内容的提问来源于stack exchange，提问作者Jyoti Prakash Mallick