JMeter多并发WebSocket测试跨线程连接异常解决方法

之前写的连接缓存逻辑用了固定key的全局属性存储，本质是整个JMeter实例里这个key只能存1个连接对象。单并发跑的时候只有1个连接，存了再取不会出错；开2并发后，两个线程先后往同一个key里写连接，后写的直接把前一个覆盖掉，最后读的时候两个线程只能拿到最后存的那一个连接，自然会出现抢连接、会话串号、一个线程读到数据另一个线程响应为空的问题。
另外之前用跨线程通信队列传sessionId的方式，没法保证取sessionId的线程和创建session的线程一一对应，队列FIFO的特性只要出现一点线程执行顺序的抖动，就会拿错sessionId，进一步加剧匹配错乱的问题。

核心思路是给每个并发线程的连接、sessionId绑上唯一的线程标识，存的时候按标识分开存，取的时候按当前线程的标识找对应的实例，从根源上避免覆盖、抢用。

1. 改造所有连接、sessionId的缓存逻辑 #

JMeter里相同线程数配置下，不同线程组里相同序号的线程可以通过内置的线程编号做绑定，这个编号通过ctx.getThreadNum()就能拿到，4个线程组里的第0个、第1个线程编号是一一对应的，直接用这个编号当缓存key的后缀即可。

第一步：Presenter建连环节改造 #

在Presenter建连（原测试流程第1步）的WebSocket采样器下加JSR223 PostProcessor，替换原来固定key存连接的代码：

// 获取当前线程的编号，作为绑定匹配的唯一标识
def threadNum = ctx.getThreadNum()
// 获取当前线程建连成功的连接实例
def presenterConn = sampler.threadLocalCachedConnection.get()
// 用带线程编号的key存连接，不同线程的key不一样，不会互相覆盖
props.put("presenter_conn_${threadNum}", presenterConn)
// 把当前线程创建的sessionId也按同样规则存好，给后面Participant订阅用
// 注意把这里的sessionId变量名替换为实际提取存储的变量名
def currentSessionId = vars.get("session_id")
props.put("match_session_id_${threadNum}", currentSessionId)

第二步：Participant建连环节改造 #

首先删掉原来从队列里取sessionId的逻辑，在Participant建连（原测试流程第2步）的WebSocket采样器下加JSR223 PreProcessor，拿当前线程编号匹配对应的sessionId用来订阅：

def threadNum = ctx.getThreadNum()
// 只获取和当前线程编号绑定的sessionId，不会拿错其他线程的
def targetSessionId = props.get("match_session_id_${threadNum}")
// 把sessionId存到当前线程的变量里，供采样器订阅使用
vars.put("target_session_id", targetSessionId)

等Participant建连、订阅成功后，在这个采样器下再加一个JSR223 PostProcessor，缓存Participant侧的连接：

def threadNum = ctx.getThreadNum()
def participantConn = sampler.threadLocalCachedConnection.get()
props.put("participant_conn_${threadNum}", participantConn)

第三步：Presenter发指令环节改造 #

在原测试流程第3步的Presenter发指令采样器下加JSR223 PreProcessor，绑定当前线程对应的Presenter连接，不要用全局固定的连接：

def threadNum = ctx.getThreadNum()
def matchedConn = props.get("presenter_conn_${threadNum}")
sampler.threadLocalCachedConnection.set(matchedConn)

第四步：Participant读消息环节改造 #

在原测试流程第4步的WebSocket Single Read Sampler下加JSR223 PreProcessor，绑定当前线程对应的Participant连接，替换原来固定key取连接的代码：

def threadNum = ctx.getThreadNum()
def matchedConn = props.get("participant_conn_${threadNum}")
sampler.threadLocalCachedConnection.set(matchedConn)

2. 移除原有错误的缓存、传值逻辑 #

• 删掉原来所有用presenterConnection、Participant1Connection这种固定key存、取连接的代码
• 删掉原来加的Inter-Thread Communication PostProcessor队列传sessionId的配置，这部分逻辑已经被按线程编号绑定的方式替代，留着反而会导致值错乱

3. 线程组配置对齐，避免顺序错乱 #

• 4个线程组的线程数必须配置成完全一样的值，不要出现某个线程组线程数和其他不一致的情况
• 每个线程组都勾选Delay Thread creation until needed选项，不要提前创建所有线程导致编号错位
• 如果测试步骤有严格的先后顺序要求，可以在测试计划里加同步定时器，保证前一个步骤的所有线程全部执行完成后，再启动下一个步骤的线程，避免线程还没建完连后面步骤就开始读，拿不到对应连接。同步点设置如下：
  • 第1步所有Presenter建连存完连接、sessionId后，再放行第2步的Participant线程
  • 第2步所有Participant建连订阅、存完连接后，再放行第3步的发指令线程
  • 第3步所有指令发完后，再放行第4步的读消息线程

调试阶段可以在每个绑定连接的JSR223脚本里加一行日志打印，输出当前线程编号、绑定的sessionId、连接对象的哈希值，执行2并发测试时检查：

• 相同编号的Presenter、Participant线程拿到的sessionId完全一致
• 两个线程拿到的连接实例哈希值不一样，不存在两个线程共用同一个连接的情况
• 最后读消息环节两个线程都能拿到对应会话的响应，不会出现空响应、串消息的问题

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