使用 akka-http 的不同执行上下文和线程分配答案

【问题标题】：Different execution contexts and thread allocation with akka-http使用 akka-http 的不同执行上下文和线程分配
【发布时间】：2016-07-06 11:53:43
【问题描述】：

不确定是否适合这个问题的 stackexchange

我有一个 akka-http 应用程序，它充当一些繁重计算的前端。它处理的请求在处理它们所需的时间上有所不同。有些在一秒钟内完成，有些则需要更多时间。计算是纯异步的，在任何时候都没有Await，我用 Future 完成请求，即：

val spotsJsonF: Future[String] = spotsF.map(spots => DebugFormatter.produceJson(text, spots._1, spots._2, env))

complete(spotsJsonF.map { t => HttpEntity(ContentTypes.`application/json`, t) })

我的要求/假设：

我需要最大限度地提高并行性，即不应在重负载下拒绝连接
如果服务繁忙，我可以忍受一些（甚至很小的）请求需要更长时间
我可以忍受一些超长请求在重负载下超时，只要它们在 HTTP 请求超时完成后不会过多影响并行性。

为此，我为繁重的计算提供了一个单独的执行上下文（即 Scala 的默认 ExecutionContext.global），即它在与 Akka http 调度程序使用的线程池不同的线程池上生成和修改 Futures。我认为这将停止计算“坐在” Akka 的线程上，因此它可以接受更多连接。目前它是 Akka 的默认调度程序（我的 reference.conf 为空）：

    "default-dispatcher": {
      "attempt-teamwork": "on",
      "default-executor": {
        "fallback": "fork-join-executor"
      },
      "executor": "default-executor",
      "fork-join-executor": {
        "parallelism-factor": 3,
        "parallelism-max": 64,
        "parallelism-min": 8,
        "task-peeking-mode": "FIFO"
      },
      "mailbox-requirement": "",
      "shutdown-timeout": "1s",
      "thread-pool-executor": {
        "allow-core-timeout": "on",
        "core-pool-size-factor": 3,
        "core-pool-size-max": 64,
        "core-pool-size-min": 8,
        "fixed-pool-size": "off",
        "keep-alive-time": "60s",
        "max-pool-size-factor": 3,
        "max-pool-size-max": 64,
        "max-pool-size-min": 8,
        "task-queue-size": -1,
        "task-queue-type": "linked"
      },
      "throughput": 5,
      "throughput-deadline-time": "0ms",
      "type": "Dispatcher"
    },

但是发生的情况是，在 Akka 由于超时而取消请求后，长时间运行的计算会继续执行很长时间。由于内核数量有限，这意味着即使不再需要启动此过载的计算，被拒绝的请求的数量也会开始增加。

显然，我不知道如何正确管理此应用程序中的线程。

满足我的要求的最佳方式是什么？几个线程池 - 好/坏主意？我需要明确取消吗？可能目前使用 Scala 的原版 Future 不是最佳选择？

【问题讨论】：

您是否考虑过创建单独的演员，每个资源一个？扩展它应该很容易，将繁重的计算代理到其他节点，并且主节点在这种情况下仅用作路由器。
我们不将其用作分布式应用程序，因为它必须在内存中保持非常重 (13G) 的结构，因此我们仅将其部署在大型 EC2 实例上

标签： multithreading scala akka akka-http

【解决方案1】：

对我来说，这听起来并不是关于管理线程，将繁重的工作隔离到您已经完成的单独调度程序，而是关于管理实际处理。

为了能够在工作中停止长时间运行的进程，可以这么说，您需要将其拆分为更小的块，以便在不再需要时中止它。

Actor 的一个常见模式是让处理 Actor 存储“到目前为止”的结果或将其作为消息发送给自己，这样它就可以对介于两者之间的“停止工作”消息做出反应，或者可能检查它是否已经处理了很长时间，它应该中止。例如，触发工作负载的消息可以包含这样一个超时值，以允许“客户端”指定它。

（这与在手动线程和阻塞应用程序中正确处理InterruptedException 和Thread.isInterrupted 基本相同）

【讨论】：