scala – 如何在Future和Thread.sleep示例中使用spray.io实现高

发布时间：2020-12-16 19:21:47 所属栏目：安全来源：网络整理

导读：我正在尝试以下POC来检查如何获得高并发性 implicit def executionContext = context.system.dispatchers.lookup("async-futures-dispatcher") implicit val timeout = 10 seconds val contestroute = "/contestroute" { get { respondWithMediaType(`applic

我正在尝试以下POC来检查如何获得高并发性

implicit def executionContext = context.system.dispatchers.lookup("async-futures-dispatcher")
  implicit val timeout = 10 seconds


  val contestroute = "/contestroute" {
      get {
          respondWithMediaType(`application/json`) {
            dynamic {
                onSuccess(
                  Future {
                    val start = System.currentTimeMillis()
                    // np here should be dealt by 200 threads defined below,so why 
                    // overall time takes so long? why doesn't it really  utilize all
                    // threads I have given to it? how to update the code so it 
                    // utilizes the 200 threads?
                    Thread.sleep(5000) 
                    val status = s"timediff ${System.currentTimeMillis() - start}ms ${Thread.currentThread().getName}"
                    status
                  })  { time =>
                  complete(s"status: $time")
                }
            }
          }
      }
  }

我的配置：

async-futures-dispatcher {
  # Dispatcher is the name of the event-based dispatcher
  type = Dispatcher
  # What kind of ExecutionService to use
  executor = "thread-pool-executor"
  # Configuration for the thread pool
  thread-pool-executor {
    # minimum number of threads to cap factor-based core number to
    core-pool-size-min = 200
    # No of core threads ... ceil(available processors * factor)
    core-pool-size-factor = 20.0
    # maximum number of threads to cap factor-based number to
    core-pool-size-max = 200
  }
  # Throughput defines the maximum number of messages to be
  # processed per actor before the thread jumps to the next actor.
  # Set to 1 for as fair as possible.
  throughput = 100
}

但是当我像这样运行apache bench时：

ab -n 200 -c 50 http://LAP:8080/contestroute

我得到的结果是：

Server Software:        Apache-Coyote/1.1
Server Port:erred:      37500 bytes
HTML transferred:       10350 bytes
Requests per second:    4.31 [#/sec] (mean)
Time per request:       34776.278 [ms] (mean)
Time per request:       231.842 [ms] (mean,across all concurrent requests)
Transfer rate:          1.05 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        5  406 1021.3      7    3001
Processing: 30132 30466 390.8  30308   31231
Waiting:    30131 30464 391.8  30306   31231
Total:      30140 30872 998.9  30353   33228            8080

Document Path:          /contestroute
Document Length:        69 bytes

Concurrency Level:      150
Time taken for tests:   34.776 seconds
Complete requests:      150
Failed requests:        0
Write errors:           0
Non-2xx responses:      150
Total transferred:      37500 bytes
HTML transferred:       10350 bytes
Requests per second:    4.31 [#/sec] (mean)
Time per request:       34776.278 [ms] (mean)
Time per request:       231.842 [ms] (mean,across all concurrent requests)
Transfer rate:          1.05 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        5  406 1021.3      7    3001
Processing: 30132 30466 390.8  30308   31231
Waiting:    30131 30464 391.8  30306   31231
Total:      30140 30872 998.9  30353   33228

我错过了一些大事吗？我需要改变什么来让我的喷雾和期货利用我给它的所有线程？

(添加我在tomcat servlet 3.0上运行)

解决方法

在您的示例中,所有喷涂操作和阻塞操作都在相同的上下文中进行.你需要拆分2个上下文：

另外我没有看到使用动态的原因,我想只是’完整’应该是好的.

implicit val timeout = 10.seconds

  // Execution Context for blocking ops
  val blockingExecutionContext = {
    ExecutionContext.fromExecutor(Executors.newFixedThreadPool(2000))
  }

  // Execution Context for Spray
  import context.dispatcher

  override def receive: Receive = runRoute(contestroute)

  val contestroute = path("contestroute") {
    get {

        complete {
          Future.apply {
            val start = System.currentTimeMillis()
            // np here should be dealt by 200 threads defined below,so why
            // overall time takes so long? why doesn't it really  utilize all
            // threads I have given to it? how to update the code so it
            // utilizes the 200 threads?
            Thread.sleep(5000)
            val status = s"timediff ${System.currentTimeMillis() - start}ms ${Thread.currentThread().getName}"
            status
          }(blockingExecutionContext)
        }

    }
  }

之后你可以测试它

ab -n 200 -c 200 http://LAP:8080/contestroute

并且你会看到喷雾将创建所有200个线程用于阻塞操作

结果：

Concurrency Level:      200
Time taken for tests:   5.096 seconds

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!