如何在Gatling gRPC负载脚本中为单个用户创建多服务器流

Gatling中gRPC serverStream的名称在单个用户会话内必须唯一，你当前固定使用stream1作为流名称，当同一个用户尝试创建第二个流时，就会触发"already exists"错误。

为每个流生成会话内唯一的名称即可解决该问题，常见的实现方式有两种：

1. 利用会话重复索引（适合批量创建固定数量的流） #

如果需要每个用户创建N个流，可以用repeat循环结合会话的repeatIndex生成唯一流名：

class StreamingSimulation {

  private final val sc = scenario("simulation-lt-scenario")
    // 每个用户创建3个独立流
    .repeat(3) {
      feed(feederFor("payload", requestHelper.getStreamingRequest()))
      // 生成唯一流名：stream-<用户ID>-<循环索引>
      .exec(session => session.set("streamName", s"stream-${session.userId}-${session.repeatIndex}"))
      .exec(
          grpc("action request").serverStream("${streamName}")
              .start(GrpcService.getMyStreamingMethod)($"payload")
              .check(extract { r: MyResponse => Option(r.someState) } is STATE_A)
      )
    }

  setUp(
    sc.inject(rampUsers(1))
      .throttle(
        reachRPS(throttleRPS) in (60 seconds),
        holdFor(300 seconds)
      )
      .maxDuration(300 seconds)
      .protocols(grpc(configureChannel(config)))
  )
}

2. 使用自定义计数器（适合动态创建流） #

如果流的数量不固定，可以在会话中维护一个计数器，每次创建流时递增并生成唯一名称：

class StreamingSimulation {

  private final val sc = scenario("simulation-lt-scenario")
    // 初始化流计数器
    .exec(session => session.set("streamCount", 0))
    // 创建第一个流
    .feed(feederFor("payload", requestHelper.getStreamingRequest()))
    .exec(session => session.set("streamName", s"stream-${session.userId}-${session("streamCount")}"))
    .exec(
        grpc("action request 1").serverStream("${streamName}")
            .start(GrpcService.getMyStreamingMethod)($"payload")
            .check(extract { r: MyResponse => Option(r.someState) } is STATE_A)
    )
    // 递增计数器
    .exec(session => session.set("streamCount", session("streamCount").as[Int] + 1))
    // 创建第二个流
    .feed(feederFor("payload", requestHelper.getStreamingRequest()))
    .exec(session => session.set("streamName", s"stream-${session.userId}-${session("streamCount")}"))
    .exec(
        grpc("action request 2").serverStream("${streamName}")
            .start(GrpcService.getMyStreamingMethod)($"payload")
            .check(extract { r: MyResponse => Option(r.someState) } is STATE_A)
    )

  setUp(
    sc.inject(rampUsers(1))
      .throttle(
        reachRPS(throttleRPS) in (60 seconds),
        holdFor(300 seconds)
      )
      .maxDuration(300 seconds)
      .protocols(grpc(configureChannel(config)))
  )
}

• 每个serverStream在会话中是独立维护的，Gatling会后台监听流的响应直到会话结束或显式关闭流。如果需要主动关闭流，可以使用.end()方法：

  .exec(grpc("close stream").serverStream("${streamName}").end())
  

• 流名称的唯一性仅针对单个用户会话，不同用户之间可以使用相同的流名称，不会产生冲突。

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