在gatling模拟期间再次加载CSV文件

问题描述

我正在尝试在两种情况下进行模拟。 设置看起来像这样

setUp(
 scn1.inject(rampConcurrentUsers(minUserCount) to (tps) during (testRampSeconds seconds),constantConcurrentUsers(tps) during (testDurationSeconds seconds)).throttle(reachRps(tps) in (1 seconds),holdFor(testDurationSeconds seconds),jumpToRps(0),holdFor(60 seconds)).protocols(httpProtocol),scn2.inject(nothingFor(testDurationSeconds),atOnceUsers(1)).protocols(httpProtocol)
  )

方案1进行REST API调用，并将响应数据保存在csv文件中

要写入文件的PrintWriter：

  val fileWriter = {
    val fos = new java.io.FileOutputStream("foo.csv")
    val appTokenWriter = new java.io.PrintWriter(fos,true)
    appTokenWriter.println("data")
    appTokenWriter
  }

方案2等待方案1完成，以便它可以立即处理csv文件中的所有条目。

.foreach(csv(foo.csv).readRecords,"record") {
      exec(flattenMapIntoAttributes("${record}"))
      .exec(session => {
        println(session("data").as[String])
        session
      })
      .exec(...REST Call...)

csv（foo.csv）.readRecord没有拾取由censing1刷新到csv中的值。 Scenario2中的csv（foo.csv）.readRecords.size = 0，但我可以看到文件中写入的数据

这是在两个方案之间维护数据的正确方法吗？ 如果是，如何使.readRecords API选择最新更改。 如果否，请提出正确的方法。

编辑：基于Stephane的实现，尝试使用ConcurrentLinkedDeque

package com.intuit.platform.identity.oauth2.performance.simulation

import java.util.concurrent.ConcurrentLinkedDeque

import io.gatling.core.Predef._
import io.gatling.core.Feeder.Feeder
import io.gatling.http.Predef.http

import scala.concurrent.duration._
import scala.util.Random

/**
 * @author vnarayanan
 */
class SOSimulation extends Simulation {

  val httpProtocol = http.baseUrl("localhost:8080")

  val dataQueue = new ConcurrentLinkedDeque[String]()

  val saveInQueue = exec{ session => {
    dataQueue.offer(session("data").as[String])
    session
  }
  }

  class datafeeder extends Feeder[String] {
    override def hasNext: Boolean = dataQueue.size() > 0
    override def next(): Map[String,String] = {
      Map("data" ->dataQueue.poll())
    }
  }
  
  val scenario1  = scenario("Scenario1")
    .exec(session => {
      session.set("data",Random.alphanumeric.take(20).mkString)
    })
    .exec(saveInQueue)

//I need to stop the requests when the queue is empty. tried it with two solutions:
//  1) asLongAs
//  2) repeat (commented)


  val scenario2  = scenario("Scenario2")
    .exec(session => {
      println(dataQueue.size())
      session
    })
    .asLongAs(dataQueue.size() > 0) {
      exec(session => {
        println(dataQueue.size())
        session
      })
        .Feed(new datafeeder())
        .exec(session => {
          println(session("data").as[String])
          session
        })
    }

//  With Repeat

//  val scenario2  = scenario("Scenario2")
//    .exec(session => {
//      println(dataQueue.size())
//      session
//    })
//    .repeat(dataQueue.size() ) {
//      exec(session => {
//        println(dataQueue.size())
//        session
//      })
//        .Feed(new datafeeder())
//        .exec(session => {
//          println(session("data").as[String])
//          session
//        })
//    }
    

  setUp(
    scenario1.inject(atOnceUsers(10)).throttle(reachRps(1) in (1 seconds),holdFor(5 seconds),holdFor(10 seconds)).protocols(httpProtocol),scenario2.inject(nothingFor(5 seconds),atOnceUsers(1)).protocols(httpProtocol)

  ).maxDuration(10 seconds)
}

解决方法

这是我们内置的csv馈送器无法实现的。 现在，请记住Feeder是Iterator[Map[String,Any]]的别名，因此您可以编写自己的实现以支持用例。 例如，您可以将数据保留在ConcurrentLinkedDeque的内存中，而不是写入文件或从文件中读取。