将MQTT连接作为属性传递时，订阅与断开连接功能失效问题

问题详情 #

• 测试场景：MQTT消息测试，需在多线程组间共享单个MQTT连接
• 当前实现方式：通过JMeter属性保存和读取连接对象，代码如下：

  props.put('connection', vars.getObject('conn'))
  vars.putObject('conn', props.get('connection'))
  

• 异常现象：
  • Publish功能完全正常，但Subscribe和Disconnect功能失效
  • 订阅操作报错栈：

    at java.util.concurrent.ConcurrentHashMap.get(ConcurrentHashMap.java:936) ~[?:?]
    at net.xmeter.samplers.SubSampler.sample(SubSampler.java:136) ~[mqtt-xmeter-2.0.2-jar-with-dependencies.jar:?]
    at org.apache.jmeter.threads.JMeterThread.doSampling(JMeterThread.java:651) ~[ApacheJMeter_core.jar:5.5]
    at org.apache.jmeter.threads.JMeterThread.executeSamplePackage(JMeterThread.java:570) ~[ApacheJMeter_core.jar:5.5]
    at org.apache.jmeter.threads.JMeterThread.processSampler(JMeterThread.java:501) ~[ApacheJMeter_core.jar:5.5]
    at org.apache.jmeter.threads.JMeterThread.run(JMeterThread.java:268) ~[ApacheJMeter_core.jar:5.5]
    at java.lang.Thread.run(Thread.java:833) ~[?:?]
    

  • 断开连接报错：响应码:501 响应信息:断开连接失败

问题原因 #

JMeter的props（全局属性）直接存储的对象未经过序列化处理，跨线程组传递MQTT连接这类线程敏感的有状态对象时，会破坏连接的线程绑定上下文。Xmeter的MQTT采样器在执行Subscribe/Disconnect操作时，依赖连接对象的并发安全访问机制，直接传递会导致ConcurrentHashMap访问异常，以及断开连接时无法识别有效连接实例。

解决方法 #

方法1：序列化后通过属性传递连接对象 #

通过序列化将连接对象转为Base64字符串存储到属性，读取时反序列化恢复，确保对象完整性：

• 保存连接到全局属性（在连接建立的线程组中执行）：

  import java.io.ByteArrayOutputStream;
  import java.io.ObjectOutputStream;
  import org.apache.commons.codec.binary.Base64;
  
  ByteArrayOutputStream bos = new ByteArrayOutputStream();
  ObjectOutputStream oos = new ObjectOutputStream(bos);
  oos.writeObject(vars.getObject('conn'));
  oos.flush();
  props.put('connection', Base64.encodeBase64String(bos.toByteArray()));
  

• 从全局属性读取连接（在其他线程组中执行）：

  import java.io.ByteArrayInputStream;
  import java.io.ObjectInputStream;
  import org.apache.commons.codec.binary.Base64;
  
  byte[] data = Base64.decodeBase64(props.get('connection').toString());
  ByteArrayInputStream bis = new ByteArrayInputStream(data);
  ObjectInputStream ois = new ObjectInputStream(bis);
  vars.putObject('conn', ois.readObject());
  

方法2：使用Test Plan级BeanShell统一管理连接 #

在Test Plan添加BeanShell PreProcessor，提前建立MQTT连接并序列化存储到全局属性，所有线程组直接从属性读取恢复后的连接对象，避免手动在每个线程组中处理传递逻辑。

方法3：添加同步锁确保线程安全 #

若MQTT客户端不支持多线程并发操作，在执行Subscribe/Disconnect的采样器中添加同步锁，避免并发访问冲突：

synchronized(vars.getObject('conn')) {
    // 这里执行Subscribe或Disconnect操作
}

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