java - 通过 toStrict() 将 HttpEntity 转换为 String 后导致延迟的原因

Question

我将 akka-http 用于响应时间目标非常短的 Web 应用程序。我将路由 DSL completeWithFuture 方法与 CompletableFuture 链一起使用。

我注意到，当使用 CompletionStage 方法的 XXXasync 变体链接每个 future 并传递相同的执行器时，用于处理阶段的线程可以任意更改，导致某些请求的响应时间更长，以防所有线程使用指定的执行器。因此，我将我的自定义执行程序传递给第一个 CompletableFuture，并将所有后续阶段与普通变体链接在一起，以便为它们使用相同的线程。

问题:一个阶段通过 HttpEntity.toStrict() 将 HttpEntity 转换为字符串，该方法使用来自 akka.actor 的线程。默认调度程序。随着工作量的增加，越来越多的请求在下一阶段开始时超过了所需的响应时间，尽管将超时传递给 toStrict 远低于目标响应时间并且没有看到超时异常.

简化代码:

private Route handleRequest(final HttpRequest request) {
    return completeWithFuture(CompletableFuture.runAsync(() -> preprocessing(), systemDispatcher) // Dispatcher 1
            .thenCompose((preprocessingResult) -> // please ignore that preprocessingResult is not used in that simplified version
            entityToString(request.entity()).thenApply((requestString) -> generateResponse(requestString))));
}

public CompletionStage<String> entityToString(final HttpEntity entity) {
        long start = System.nanoTime();
        return entity.toStrict(bodyReadTimeoutMillis, materializer).thenApplyAsync((final Strict strict) -> {
            System.out.println(start-System.nanoTime()); // varies between <1ms and >500ms
            return strict.getData().utf8String();
        }, systemDispatcher); // Dispatcher 2
    }

所以我的猜测是从我的自定义执行程序的线程切换到 akkas 默认 actor 调度程序线程之一并返回导致了问题。

问题:对于我的 entityToString 方法的延迟是否有其他解释？有没有一种方法可以实现与 toStrict 相同的效果，即将整个可能分块的消息正文作为字符串获取，同时避免多次切换线程？

请注意，我需要 toStrict 方法的超时功能来中止对慢速 POST 请求的处理。

更新

前几天想了想，想实现akka保证的非阻塞读，不切换线程是不可能的。所以真正的问题是明显的高延迟可能是由 toStrict 之后的调度引起的。

我尝试使用不同的调度程序(请参阅上面代码中的注释 Dispatcher 1/Dispatcher 2)并在延迟的情况下记录 Dispatcher 2 的居民计数超过 50ms。我找不到合适的文档，但我认为这是计划任务的数量。 我用 10000 个请求、200 个并发连接运行 apache bench，并有 55 次延迟超过 50 毫秒。输出显示最多 80 个居民。

我在 Amazon 的 m3.2xlarge 实例(8 个 vCPU、30GB RAM、Ubuntu 16.04)上运行了该测试，没有其他进程消耗任何明显的 CPU 量。调度程序的类型为 fork-join-executor，parallelism-factor = 1。

具有更多可变数量的并发请求的实际流量会导致超过限制的请求率不断增加(高达 50%)。

处理请求的平均时间低于 1 毫秒。是什么导致了 toStrict 之后的罕见延迟以及如何避免它？

Answer 1

您可以尝试使用 akka-streams 获取实体内容，并以不同的方式实现超时要求，而不是实体读取。

entity.getDataBytes().runFold(ByteString.empty(), ByteString::concat, materializer)
.thenCompose(r -> r.utf8String());