咨询Gatling对POST端点事件流的断言与首消息性能测试方案（含JMeter建议）

Gatling 实现方案 #

Gatling官方的SSE组件确实仅支持GET请求，但可以通过自定义HTTP请求结合流处理适配POST类型的事件流接口，核心思路是直接处理分块编码的响应流，提取第一条消息后主动中断连接，无需等待整个流结束。

核心实现步骤 #

1. 配置请求头：设置Accept: text/event-stream以及POST请求所需的Content-Type（如application/json）。
2. 自定义流处理逻辑：通过bodyStream.transform读取响应流，捕获第一条以data:开头的消息，完成后关闭流终止连接。
3. 记录第一条消息的响应时间：通过Session变量记录请求发送前的时间戳，拿到消息后计算时间差，替代默认的全流响应时间。

代码示例 #

import io.gatling.core.Predef._
import io.gatling.http.Predef._
import java.io.{BufferedReader, InputStreamReader}

val scn = scenario("POST Event Stream Test")
  .exec(session => {
    // 记录请求开始时间
    session.set("requestStartTime", System.currentTimeMillis())
  })
  .exec(
    http("POST Event Stream Request")
      .post("/your-event-stream-endpoint")
      .header("Accept", "text/event-stream")
      .header("Content-Type", "application/json")
      .body(StringBody("""{"param1": "value1"}""")) // 替换为实际请求参数
      .check(
        bodyStream.transform { inputStream =>
          val reader = new BufferedReader(new InputStreamReader(inputStream))
          var line: String = null
          var firstMessage: String = null
          // 读取流直到拿到第一条data消息
          while (firstMessage == null && (line = reader.readLine()) != null) {
            if (line.startsWith("data:")) {
              firstMessage = line.substring(5).trim()
              // 主动关闭流，终止连接
              inputStream.close()
            }
          }
          Option(firstMessage)
        }.saveAs("firstMessage")
      )
  )
  .exec(session => {
    // 计算第一条消息的响应时间
    val startTime = session("requestStartTime").as[Long]
    val currentTime = System.currentTimeMillis()
    val firstMsgTime = currentTime - startTime
    session.set("firstMsgResponseTime", firstMsgTime)
  })
  .exec(session => {
    // 断言第一条消息内容
    val expectedMsg = """{"message": "first message"}"""
    val actualMsg = session("firstMessage").as[String]
    if (actualMsg != expectedMsg) {
      throw new AssertionError(s"First message mismatch: expected $expectedMsg, got $actualMsg")
    }
    session
  })

setUp(scn.inject(atOnceUsers(1))).protocols(http.baseUrl("http://your-service-host"))

JMeter 实现方案 #

JMeter通过开启响应流模式，结合自定义脚本可实现相同需求，无需等待事件流结束即可捕获第一条消息并完成性能统计与断言。

核心实现步骤 #

1. 配置HTTP请求：设置POST方法、目标端点，添加HTTP Header Manager配置Accept: text/event-stream和请求对应的Content-Type。
2. 开启响应流：在HTTP请求的「Advanced」标签下勾选「Response Streaming」，避免缓存整个响应。
3. 记录开始时间：添加JSR223 PreProcessor记录请求发起前的时间戳。
4. 处理流并断言：添加JSR223 PostProcessor读取响应流，捕获第一条消息，计算时间差，覆盖默认响应时间并完成断言。

脚本示例 #

1. JSR223 PreProcessor（记录开始时间）

vars.put("START_TIME", String.valueOf(System.currentTimeMillis()))

2. JSR223 PostProcessor（处理流、计算时间、断言）

long startTime = vars.getLong("START_TIME")
def reader = new BufferedReader(new InputStreamReader(sampleResult.getResponseDataStream()))
String line
String firstMsg = null

while ((line = reader.readLine()) != null) {
    if (line.startsWith("data:")) {
        firstMsg = line.substring(5).trim()
        long elapsed = System.currentTimeMillis() - startTime
        // 覆盖采样器的响应时间为第一条消息的耗时
        sampleResult.setResponseTime(elapsed)
        vars.put("FIRST_MSG_TIME", String.valueOf(elapsed))
        reader.close()
        break
    }
}

// 断言第一条消息内容
def expectedMsg = """{"message": "first message"}"""
if (firstMsg == null) {
    sampleResult.setSuccessful(false)
    sampleResult.setResponseMessage("No first message received from event stream")
} else if (!firstMsg.equals(expectedMsg)) {
    sampleResult.setSuccessful(false)
    sampleResult.setResponseMessage("First message mismatch: expected '$expectedMsg', got '$firstMsg'")
}

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