如何使用JMeter实现跨IBM MQ队列的异步消息性能测试与报告生成

1. 消息ID匹配与线程间通信实现 #

线程间通信函数调用方法 #

你无法在JSR223 Sampler中直接写${__fifoPop()}这类函数表达式，可通过JMeter上下文的函数执行接口调用插件能力，生产侧写入FIFO的代码示例如下：

// 生成唯一消息ID，可写入消息属性或payload中
def msgId = UUID.randomUUID().toString()
def sendTime = System.currentTimeMillis()
// 原有写入MQ1的业务逻辑
// ...

// 将消息ID和发送时间拼接后写入FIFO队列
ctx.getFunctions().executeFunction("__fifoPut", ["msg_pending_queue", "${msgId}|${sendTime}"] as String[])
SampleResult.setSuccessful(true)

匹配消费实现 #

推荐优先使用消息选择器实现精准消费，无需遍历队列，效率更高：

// 从FIFO取出待匹配的消息信息
def fifoContent = ctx.getFunctions().executeFunction("__fifoPop", ["msg_pending_queue", ""] as String[])
if (!fifoContent) {
    SampleResult.setSuccessful(false)
    SampleResult.setResponseMessage("FIFO队列无待匹配消息")
    return
}
// 拆分获取消息ID和发送时间戳
def (targetMsgId, sendTime) = fifoContent.split("\\|")
sendTime = sendTime.toLong()

// 创建带选择器的消费者，仅接收匹配消息ID的消息
// 注意：custom_msg_id为生产侧写入消息时自定义的属性名，需和生产逻辑对应
def consumer = session.createConsumer(outboundMqQueue, "custom_msg_id = '" + targetMsgId + "'")
def receivedMsg = consumer.receive(4000) // 超时时间设为4秒
consumer.close()

if (receivedMsg == null) {
    SampleResult.setSuccessful(false)
    SampleResult.setResponseMessage("超时：未找到ID为${targetMsgId}的匹配消息")
    return
}

如果需要仅浏览队列不消费做全量匹配，可使用JMS的QueueBrowser接口遍历所有消息，匹配到目标ID后再执行消费操作，适合无法使用消息选择器的场景。

2. 处理时间计算 #

消费到匹配消息后，直接用当前时间戳减去生产侧记录的发送时间戳即可得到处理耗时，将耗时赋值给采样器的响应时间字段即可被JMeter统计：

def processTime = System.currentTimeMillis() - sendTime
// 将处理耗时设置为采样器的响应时间
SampleResult.setResponseTime(processTime)
SampleResult.setSuccessful(true)
SampleResult.setResponseMessage("消息${targetMsgId}处理完成，耗时${processTime}ms")

3. 聚合报告生成 #

无需自定义统计逻辑，直接添加JMeter原生的Aggregate Report监听器即可，它会自动收集所有消费采样器的结果，生成你需要的平均耗时、各百分位耗时、吞吐量（每秒事务数）、错误率等全部指标，和普通接口测试的聚合报告逻辑完全一致。

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