1. 项目概述当JMeter遇上SSE一场“持久”的较量如果你正在用JMeter测试一个实时数据推送的接口比如股票行情、聊天消息或者后台任务进度那么你很可能已经接触过SSEServer-Sent Events服务器发送事件。这玩意儿用起来是真方便一个HTTP长连接服务器就能源源不断地把数据“流”过来。但当你兴冲冲地打开JMeter准备给它来个压力测试时十有八九会掉进坑里。你会发现脚本要么收不到完整数据要么内存暴涨把机器搞崩要么结果树里一片混乱根本没法断言。这感觉就像你开着一辆F1赛车去跑越野赛道引擎轰鸣但寸步难行。“避坑指南JMeter处理SSE响应时常见的5个错误及解决方案”这个标题精准地戳中了性能测试工程师和开发者在做SSE接口压测时的痛点。它不是一个泛泛而谈的教程而是直指那些让你调试到怀疑人生的具体问题。SSE协议本身并不复杂但JMeter作为一个基于HTTP/1.1短连接模型设计的工具其默认的请求-响应生命周期与SSE的长连接、流式响应特性存在根本性的冲突。这篇文章的目的就是帮你捋清这些冲突点把常见的五个大坑——从连接建立、数据接收、资源管理到结果分析——一个个填平让你手里的JMeter能从“越野模式”切换回“赛道模式”真正驾驭SSE这种流式接口的测试。2. 核心原理拆解为什么JMeter“看不懂”SSE在开始填坑之前我们必须先搞清楚JMeter和SSE之间到底哪里“不对付”。不理解这个所有的解决方案都只是死记硬背的步骤换个场景可能又失灵了。2.1 SSE协议的核心机制SSE本质上是一个基于HTTP的长连接。客户端发起一个普通的GET请求但服务器在响应时会做两件关键的事设置响应头Content-Type: text/event-stream和Cache-Control: no-cache。这是告诉浏览器或客户端“嘿这不是一个一次性文档而是一个事件流别缓存一直听着。”保持连接并流式发送数据服务器不会立即关闭连接而是会以特定的格式data: {消息内容}\n\n持续地向客户端发送数据块每个数据块以两个换行符\n\n分隔。连接会一直保持直到服务器主动关闭或客户端断开。2.2 JMeter默认行为与SSE的冲突JMeter的线程模型是为模拟大量独立、短暂的HTTP请求而优化的。一个HTTP Request采样器Sampler的生命周期是建立连接 - 发送请求 - 接收响应直到服务器关闭输出流- 断开连接 - 记录结果。这里就产生了几个根本矛盾冲突一响应结束的判定。JMeter认为服务器关闭输出流TCP连接才代表“响应结束”。但SSE连接是常开的服务器永远不会主动关闭除非出错或任务完成。这导致JMeter的采样器会一直“挂起”等待一个永远不会到来的结束信号最终超时。冲突二响应数据的处理。JMeter默认将整个响应体作为一个完整的字符串在请求结束后一次性交给后置处理器如正则表达式提取器或断言。但对于SSE数据是片段化、持续到达的。JMeter的默认机制无法实时处理这些片段。冲突三资源与线程占用。一个挂起的采样器会独占一个JMeter线程。如果你模拟100个并发用户就意味着100个线程被无限期挂起无法进行后续迭代这完全违背了压测的初衷。简单来说JMeter期待的是“一问一答答完就走”的对话而SSE提供的是“一问多答答个不停”的广播。不解决这个根本矛盾任何测试都无法进行。注意市面上有些文章会教你用“循环控制器短超时”来模拟这本质上是轮询Polling不是真正的SSE长连接测试无法模拟真实的长连接资源占用和服务器推送行为测试结果会失真。3. 错误一采样器无限挂起与超时这是你遇到的第一个也是最明显的错误。脚本一运行JMeter就卡住了视图里一堆采样器一直显示绿色运行中直到你设置的超时时间比如60秒到了它们才批量失败报告Response timeout。3.1 错误现象与根因分析现象在“查看结果树”中采样器状态长时间为“正在运行”无响应数据。最终因超时而失败。根因如上所述JMeter在等待响应结束但SSE连接永不结束。HTTP Request采样器的默认行为不适应流式响应。3.2 解决方案使用SSE Sampler插件最正统、最有效的解决方案是使用专为SSE设计的第三方插件。SSE Sampler插件通常指jmeter-sse或WebSocket/SSE Sampler改变了采样器的行为逻辑。实操步骤安装插件访问 JMeter 插件管理器网站如https://jmeter-plugins.org/搜索 “SSE” 或 “WebSocket”。找到合适的SSE采样器插件例如WebSocket Samplers by Peter Doornbosch这个插件包通常包含SSE支持。下载plugins-manager.jar并放入JMeter的lib/ext目录重启JMeter后通过插件管理器安装。或者直接下载插件的JAR文件放入lib/ext目录后重启JMeter。配置SSE采样器在测试计划中右键添加 - 取样器 - 你应该能看到新增的SSE Sampler或类似名称的采样器。关键配置项Server URL你的SSE接口地址例如http://localhost:8080/events。Read Timeout读取超时。这不再是等待连接关闭的超时而是等待下一条消息的超时。如果服务器超过这个时间没有发送新消息采样器会认为本次“读取”结束但连接可能仍保持。可以设置得较长如30000毫秒。Connection Timeout连接建立超时。Implementation选择HTTP。Streaming Connection确保勾选通常是默认的这告诉采样器这是一个流式连接。处理持续响应SSE采样器会持续读取消息直到达到Read Timeout或通过后置处理器主动停止。你可以在采样器下添加后置处理器如BeanShell或JSR223来实时处理每条收到的消息并进行计数、断言或存储。避坑心得插件兼容性注意JMeter版本与插件版本的兼容性。建议在测试环境先验证插件的基本功能。超时理解区分Read Timeout和Response Timeout。前者是消息间等待后者是整个请求等待。使用SSE采样器后你主要关注Read Timeout。资源清理在测试计划最后确保添加一个测试活动-SSE End采样器如果插件提供或在Teardown线程组中强制断开连接防止连接泄漏。4. 错误二响应数据不完整或截断即使用上了SSE采样器或者你通过一些“黑魔法”让普通HTTP采样器收到了数据你可能会发现“查看结果树”里显示的响应数据只有第一条或者只有一部分后面的消息不见了。4.1 错误现象与根因分析现象结果树中采样器的响应数据只显示了SSE流中的第一条或前几条消息后续消息丢失。根因普通HTTP采样器如前所述它只记录“响应结束时”缓冲区里的数据。对于SSE它可能只捕获了连接建立后服务器立即发送的第一条欢迎消息或初始数据。SSE采样器配置不当即使使用SSE采样器如果其内部的消息缓冲区大小有限或者在采样器“结束”前没有正确读取并存储所有消息也会导致数据丢失。JMeter的默认数据存储机制JMeter默认只保存采样器的“响应数据”对于持续到达的数据流它没有内置的、持续的追加存储逻辑。4.2 解决方案实时监听与存储我们需要改变思路不从采样器的“结果”里拿完整数据而是在数据到达的瞬间就把它捕获并存下来。实操步骤以SSE采样器为例添加JSR223监听器或后置处理器在SSE采样器下添加一个 - 监听器 -JSR223 Listener。选择语言推荐Groovy性能好。这个监听器会在每次收到消息时被触发。编写消息处理脚本// 获取当前采样器实例 def sampler sampler // 获取本次读取到的响应数据一条SSE消息 def response prev.getResponseDataAsString() // 简单的日志输出可在jmeter.log中查看 log.info(收到SSE消息: response) // 将消息存储到JMeter变量中供后续采样器使用例如存储到数组或列表 vars.putObject(lastSseMessage, response) // 或者更常见的做法将消息追加到一个文件或内存列表中 def messageList vars.getObject(sseMessageList) if (messageList null) { messageList new java.util.ArrayList() vars.putObject(sseMessageList, messageList) } messageList.add(response) // 示例对消息内容进行断言检查是否包含特定关键字 if (!response.contains(expected_keyword)) { prev.setSuccessful(false) prev.setResponseMessage(SSE消息未包含预期关键字) } // 如果你想在收到特定消息后停止这个SSE连接可以操作采样器 // if (response.contains(STOP)) { // sampler.setStopThread(true) // 谨慎使用会停止当前线程 // // 或者调用采样器提供的断开方法如果插件有提供API // }使用BeanShell PostProcessor或自定义Java代码原理类似但JSR223Groovy是更现代、性能更好的选择。BeanShell在复杂逻辑和高并发下可能成为瓶颈。避坑心得性能影响在每条消息到达时都执行脚本尤其是文件写入操作会对测试性能产生较大影响。对于高性能压测考虑将消息先存储在内存列表如ArrayList中在测试结束后或定时批量写入文件。变量作用域vars.putObject存储的对象是线程独立的。如果你需要跨线程聚合所有消息需要使用propsJMeter属性但要注意线程安全。消息去重与格式SSE消息可能包含id、event等字段。你的处理脚本需要能够解析data: actual message这种格式并处理可能的重连机制retry字段。5. 错误三内存消耗暴涨OOM这是最危险的一个错误可能直接导致JMeter客户端崩溃。现象是JMeter进程的内存使用率在监控工具或任务管理器中可见持续快速上升最终抛出java.lang.OutOfMemoryError: Java heap space错误。5.1 错误现象与根因分析现象JMeter运行一段时间后变卡最终崩溃。日志中报堆内存溢出错误。根因无限增长的响应缓冲区如果错误地使用了普通HTTP采样器并且没有设置超时或超时很长JMeter会试图将整个永不结束的响应流读入内存直到内存耗尽。不当的消息累积即使用了SSE采样器但如果你在JSR223脚本中将所有消息无限制地添加到一个全局的ArrayList或StringBuilder中内存也会随着测试时间线性增长。JMeter自身报告收集如果开启了“保存响应数据”的选项JMeter会为每个采样器存储完整的响应数据。对于SSE长连接这相当于在内存中保存一个不断增大的字符串非常危险。5.2 解决方案精细化内存管理实操步骤调整JMeter启动参数这是基础。编辑JMeter启动脚本jmeter.bat或jmeter找到HEAP设置。根据你的测试机器内存适当增加堆大小例如set HEAP-Xms2g -Xmx4g -XX:MaxMetaspaceSize512m。但这不是根本解决办法只是扩大了“水池”如果漏水太快还是会满。禁用不必要的响应数据保存在HTTP Request或SSE Sampler的配置中找到 “Save response as MD5 hash?” 或类似的选项。务必勾选这个。这样JMeter只保存一个固定长度的哈希值而不是庞大的响应体。在“查看结果树”等监听器中也避免配置保存完整的响应数据。设计智能的消息处理逻辑抽样存储不是每条消息都存。例如每收到10条消息才处理或存储1条。// 在JSR223 Listener中 def counter vars.getObject(msgCounter) ?: 0 counter vars.putObject(msgCounter, counter) if (counter % 10 0) { // 只处理第10, 20, 30...条消息 def messageList vars.getObject(sseMessageSampleList) ?: new java.util.ArrayList() messageList.add(prev.getResponseDataAsString()) vars.putObject(sseMessageSampleList, messageList) } // 其他消息直接忽略或只做极轻量的检查如计数滑动窗口只保留最近N条消息。def windowSize 100 def messageWindow vars.getObject(sseMessageWindow) ?: new java.util.LinkedList() messageWindow.add(prev.getResponseDataAsString()) if (messageWindow.size() windowSize) { messageWindow.removeFirst() } vars.putObject(sseMessageWindow, messageWindow)及时清理在收到特定结束事件或经过一定时间后主动清理存储的消息集合。写入磁盘对于需要完整日志的场景尽快将消息写入文件并清空内存中的列表。使用带缓冲的写入器减少IO次数。控制连接时长与并发数不要让单个SSE连接无限制地运行。在测试计划中使用定时器如Runtime Controller来控制SSE采样器的执行时长。合理设置并发用户数线程数。每个SSE长连接都会占用服务器和客户端资源。过高的并发本身就是内存溢出的主要原因。避坑心得监控是关键运行测试时打开JConsole或VisualVM连接到JMeter进程实时观察堆内存使用情况和GC活动。看到内存曲线只升不降就要警惕了。理解插件行为不同的SSE插件实现不同了解你用的插件是如何管理连接和缓冲区的。有些插件可能需要在采样器级别配置缓冲区大小。分离控制与压测机对于大规模SSE压测强烈建议使用分布式模式让多个JMeter从机Slave分担连接压力主机Master只负责收集聚合结果这能有效分散单机内存压力。6. 错误四断言与后置处理器失效在SSE流测试中你可能会发现精心配置的“响应断言”根本不触发或者“正则表达式提取器”什么也提取不到。这是因为这些元件的工作时机不对。6.1 错误现象与根因分析现象在SSE采样器下添加的响应断言即使服务器返回了错误格式的消息断言也显示通过。正则表达式提取器提取的变量为空。根因JMeter的标准后置处理器断言、提取器等是在采样器执行完毕后才执行的。对于SSE采样器其“执行完毕”可能是在Read Timeout超时之后。因此这些处理器只能对超时前最后一条收到的消息或者整个缓冲区的内容进行操作无法对流转过程中的每一条消息进行实时断言和提取。6.2 解决方案实时断言与提取我们需要将断言和提取的逻辑从采样器的“后置”阶段移动到消息到达的“即时”阶段。实操步骤使用JSR223 Assertion进行实时断言不要用标准的“响应断言”。在SSE采样器下添加一个 - 断言 -JSR223 Assertion。它的执行时机可以配置默认为“在采样器之后”但关键是我们可以把它和JSR223 Listener结合或者在其脚本中访问当前响应。更常见的模式是将断言逻辑直接写在JSR223 Listener脚本里如前面示例中的if (!response.contains(...))。这样每条消息到达时都会进行校验。在消息处理器中提取变量同样放弃使用“正则表达式提取器”。在JSR223 Listener中使用Groovy/Java的字符串处理功能如split(),substring(), 或正则表达式来解析消息并将需要的值存入JMeter变量。def message prev.getResponseDataAsString() // 假设消息格式为 event: price\ndata: {symbol:BTC,price:50000}\n\n def lines message.split(\\n) for (line in lines) { if (line.startsWith(data: )) { def jsonStr line.substring(6) // 去掉data: 前缀 // 简单提取实际可用JsonSlurper解析 def matcher jsonStr ~ /price:(\d)/ if (matcher.find()) { vars.put(latestPrice, matcher.group(1)) log.info(提取到价格: vars.get(latestPrice)) } } }这样提取出的变量如${latestPrice}可以被同一个线程后续的采样器比如一个下单接口使用。使用自定义采样器或插件的高级功能一些高级的SSE插件可能内置了按事件Event进行断言和字段提取的功能。仔细阅读插件的文档。避坑心得断言粒度对于SSE流断言的目标通常是“流中是否出现了符合/不符合预期的消息”而不是“整个响应的最终状态”。设计断言时要考虑这种持续性。变量生命周期在JSR223 Listener中设置的变量默认对该线程后续的采样器可见。确保你理解变量的作用域vars是线程局部变量。性能考量对每条消息都进行复杂的正则表达式匹配或JSON解析会增加CPU负担。确保你的测试机器能承受。7. 错误五测试结果分析与报告困难即使你成功发起了SSE连接收到了数据控制了内存也做了断言最后却发现在“聚合报告”或“汇总报告”里数据完全失真了。TPS每秒事务数高得离谱或者低得可怜响应时间毫无意义。7.1 错误现象与根因分析现象TPS异常高因为一个长连接可能收到了成百上千条消息JMeter可能将每条消息都错误地记录为一个独立的“事务”。响应时间异常长采样器的响应时间被记录为整个长连接的持续时间从连接到超时或手动停止可能长达几分钟甚至几小时这完全扭曲了服务器处理单个请求的真实性能。成功率失真连接建立成功就算成功即使流中后续消息全是错误的。根因JMeter的标准事务模型一个采样器一次请求一个事务与SSE的流式交互模型不匹配。我们需要重新定义什么是SSE测试中的“事务”和“响应时间”。7.2 解决方案重新定义事务与指标我们需要跳出JMeter默认的度量体系建立适合SSE的定制化指标收集和分析方法。实操步骤禁用或忽略默认的采样器计时对于SSE连接采样器本身在监听器如聚合报告中过滤掉它或者接受它的响应时间就是连接时长这个事实但在分析时明确知道这个数据不代表服务端处理性能。更关键的是我们要自己定义和测量业务指标。测量“消息到达延迟”这是SSE测试的核心指标之一从服务器发出消息到客户端JMeter收到消息的时间差。实现方法需要服务器端支持在发送的每条消息里带一个高精度的时间戳如data: {ts: 1640995200000, payload: ...}。在JMeter的JSR223 Listener中收到消息后立即获取当前系统时间与消息中的时间戳相减得到延迟。import groovy.json.JsonSlurper def message prev.getResponseDataAsString() def lines message.split(\\n) for (line in lines) { if (line.startsWith(data: )) { def jsonStr line.substring(6) def json new JsonSlurper().parseText(jsonStr) long serverTs json.ts long clientTs System.currentTimeMillis() long latency clientTs - serverTs // 将延迟存储起来供后续分析 def latencyList vars.getObject(latencyList) ?: new java.util.ArrayList() latencyList.add(latency) vars.putObject(latencyList, latencyList) // 也可以直接使用SampleResult记录一个自定义的“虚拟事务” // 但这需要更复杂的操作通常建议将latencyList在测试结束后输出到文件用其他工具分析 } }测量“连接建立成功率”与“消息丢失率”连接成功率SSE采样器自身的成功/失败状态可以很好地反映这一点。消息丢失率这需要业务逻辑配合。例如服务器发送的消息带有连续递增的ID。JMeter端统计收到的ID检查是否有间隔。// 假设消息格式为 data: {id: 123, ...} // 在JSR223 Listener中 def currentId json.id def lastId vars.getObject(lastReceivedId) ?: 0 if (currentId ! lastId 1) { // 发现丢包记录日志或增加计数器 def lossCounter vars.getObject(lossCounter) ?: 0 lossCounter (currentId - lastId - 1) vars.putObject(lossCounter, lossCounter) log.warn(消息丢失期望ID: ${lastId 1}, 实际收到ID: $currentId) } vars.putObject(lastReceivedId, currentId)使用SampleResult记录自定义事务高级如果你想在JMeter的报告中生成更直观的图表如消息延迟的分布可以创建自定义的SampleResult对象。这通常在JSR223 PostProcessor或BeanShell PostProcessor中完成但需要小心处理线程同步和资源管理复杂度较高。对于大多数场景将原始数据如每条消息的延迟写入CSV或JSON文件然后用专业的数据分析工具如Grafana, Python Pandas进行后期处理是更简单可靠的选择。避坑心得明确测试目标在开始SSE压测前就要想清楚你到底要衡量什么是服务器维持大量长连接的能力是消息推送的延迟还是消息的吞吐量不同的目标决定了不同的脚本设计和指标收集方式。结果可视化不要只盯着JMeter的聚合报告。将自定义指标延迟、丢包数写入文件用Excel、Grafana或自定义脚本生成图表才能获得更深入的洞察。时钟同步测量“消息到达延迟”要求JMeter测试机与服务器时钟高度同步最好使用NTP服务。否则测量值可能包含巨大的系统时间差失去意义。8. 实战配置与脚本框架示例理论说再多不如一个可运行的例子来得实在。下面我给出一个基于WebSocket/SSE Sampler插件假设已安装的完整测试计划框架它涵盖了连接、接收、处理、断言和资源清理的基本流程。测试计划结构测试计划 (Test Plan) ├─ 用户定义的变量 (User Defined Variables) │ ├─ SERVER_URLhttp://localhost:8080/sse/stream │ ├─ READ_TIMEOUT30000 │ └─ TEST_DURATION60 (seconds) ├─ 线程组 (Thread Group) │ ├─ 线程数: 10 │ ├─ Ramp-up: 5 │ └─ 循环次数: 1 (由Runtime Controller控制时长) │ │ │ ├─ 运行时控制器 (Runtime Controller) - 运行时: ${__P(TEST_DURATION,60)} │ │ │ │ │ ├─ SSE Sampler (名称: Connect to SSE Stream) │ │ │ ├─ Server URL: ${SERVER_URL} │ │ │ ├─ Read Timeout: ${READ_TIMEOUT} │ │ │ ├─ Implementation: HTTP │ │ │ └─ (其他参数默认) │ │ │ │ │ ├─ JSR223 Listener (名称: Process SSE Messages) [语言: Groovy] │ │ │ └─ 脚本内容: (见下方) │ │ │ │ │ └─ 响应断言 (JSR223 Assertion) [可选或直接在Listener中断言] │ │ │ └─ 查看结果树 (View Results Tree) [仅调试用正式压测建议禁用或使用简单数据写入器] │ └─ tearDown线程组 (tearDown Thread Group) [确保连接关闭] └─ SSE End Sampler (名称: Close SSE Connections) └─ (配置为关闭所有连接)核心JSR223 Listener脚本示例这个脚本实现了消息计数、延迟计算需服务器时间戳、简单断言和抽样存储。import groovy.json.JsonSlurper import java.util.concurrent.atomic.AtomicLong // 获取当前采样器结果 def sampleResult prev // 获取当前响应数据一条SSE消息 def rawMessage sampleResult.getResponseDataAsString() if (rawMessage null || rawMessage.trim().isEmpty()) { return // 忽略空消息 } // 1. 消息计数线程安全跨线程统计需用props def threadCounter vars.getObject(msgCount) ?: new AtomicLong(0) def currentCount threadCounter.incrementAndGet() vars.putObject(msgCount, threadCounter) log.debug(线程 ${ctx.getThreadNum()} 收到第 ${currentCount} 条消息) // 2. 解析SSE消息格式 (简单处理寻找data行) def lines rawMessage.split(\\n) def eventName message def dataContent def messageId null def retryTime null for (line in lines) { if (line.startsWith(event:)) { eventName line.substring(6).trim() } else if (line.startsWith(data:)) { dataContent line.substring(5).trim() } else if (line.startsWith(id:)) { messageId line.substring(3).trim() } else if (line.startsWith(retry:)) { retryTime line.substring(6).trim() } // 忽略以冒号开头的行注释和空行 } if (dataContent.isEmpty()) { return // 没有data字段可能是注释或心跳跳过处理 } // 3. 业务逻辑处理 (示例假设dataContent是JSON) try { def jsonSlurper new JsonSlurper() def data jsonSlurper.parseText(dataContent) // 3.1 计算延迟假设服务器返回了timestamp字段单位毫秒 if (data.timestamp) { long serverTime data.timestamp as Long long clientTime System.currentTimeMillis() long latency clientTime - serverTime // 存储延迟样本使用线程局部列表避免竞争 def latencyList vars.getObject(latencyList) ?: [] latencyList.add(latency) if (latencyList.size() 1000) { latencyList.remove(0) } // 滑动窗口防止内存溢出 vars.putObject(latencyList, latencyList) // 可以在这里设置阈值断言 if (latency 1000) { // 延迟超过1秒视为警告 // sampleResult.setSuccessful(false) // 谨慎使用这会影响整个采样器的成功状态 log.warn(高延迟消息: ID${messageId}, latency${latency}ms) } } // 3.2 检查消息内容断言 if (data.status data.status error) { // 业务逻辑错误可以记录失败计数 def errorCounter vars.getObject(errorCount) ?: new AtomicLong(0) errorCounter.incrementAndGet() vars.putObject(errorCount, errorCounter) log.error(收到错误消息: dataContent) } // 3.3 抽样存储每100条存1条到文件减少IO压力 if (currentCount % 100 0) { def sampleList vars.getObject(messageSamples) ?: [] sampleList.add([timestamp: System.currentTimeMillis(), id: messageId, data: data]) vars.putObject(messageSamples, sampleList) // 实际写入文件操作最好在测试结束后统一进行或使用异步写入器 } // 3.4 提取变量供后续采样器使用例如最新价格 if (data.price) { vars.put(latestPrice, data.price.toString()) } } catch (Exception e) { log.error(解析SSE消息JSON失败: rawMessage, e) // 可以考虑将解析失败视为测试失败 def parseFailCounter vars.getObject(parseFailCount) ?: new AtomicLong(0) parseFailCounter.incrementAndGet() vars.putObject(parseFailCount, parseFailCounter) } // 4. 模拟业务操作例如每收到10条消息发一个HTTP请求 if (currentCount % 10 0) { // 这里可以添加一个HTTP Request采样器但更优雅的方式是用if控制器和模块控制器 // 简单示例记录日志表示应该触发业务操作 log.info(已收到 ${currentCount} 条消息触发模拟业务操作...) // vars.put(triggerOperation, true) // 设置标志由后续逻辑判断 }后处理与结果收集在测试计划的最后添加一个JSR223 PostProcessor或使用BeanShell PostProcessor在测试结束后执行将各个线程收集的指标如平均延迟、消息总数、错误数汇总并输出到文件。// 此脚本可在测试计划的“tearDown线程组”或最后一个线程组中添加一个仅一次执行的采样器后执行 import java.util.concurrent.atomic.AtomicLong def totalMessages 0L def totalErrors 0L def allLatencies [] // 注意这里简化了跨线程收集实际高并发时props.getObject可能返回副本需要更精细的同步。 // 对于生产压测建议每个线程独立写文件或使用JMeter的聚合报告配合自定义指标。 ctx.getThreadGroup().getThreads().each { thread - def threadVars thread.getVariables() def msgCountObj threadVars.getObject(msgCount) if (msgCountObj instanceof AtomicLong) { totalMessages msgCountObj.get() } def errorCountObj threadVars.getObject(errorCount) if (errorCountObj instanceof AtomicLong) { totalErrors errorCountObj.get() } def latencies threadVars.getObject(latencyList) if (latencies instanceof List) { allLatencies.addAll(latencies) } } def avgLatency allLatencies ? (allLatencies.sum() / allLatencies.size()) : 0 def maxLatency allLatencies ? Collections.max(allLatencies) : 0 def minLatency allLatencies ? Collections.min(allLatencies) : 0 def summary SSE压力测试结果汇总: 总连接线程数: ${ctx.getThreadGroup().getNumThreads()} 总运行时间: ${props.get(TEST_DURATION)} 秒 总接收消息数: ${totalMessages} 总业务错误数: ${totalErrors} 消息错误率: ${totalMessages 0 ? String.format(%.2f, (totalErrors / totalMessages * 100)) : 0}% 延迟统计 (ms): 平均: ${String.format(%.2f, avgLatency)} 最大: ${maxLatency} 最小: ${minLatency} 样本数: ${allLatencies.size()} log.info(summary) // 将结果写入文件 def resultFile new File(sse_test_result_${System.currentTimeMillis()}.txt) resultFile.write(summary)这个框架提供了一个坚实的起点。你需要根据自己服务器的SSE接口的具体格式消息结构、事件类型和你的测试目标压测连接数、消息延迟、吞吐量来调整消息解析逻辑、断言条件和指标收集方式。记住测试SSE接口更像是在测试一个持续的服务而不是离散的API调用你的测试脚本也需要具备这种“持续性”和“状态性”的思维。