当前位置: 首页 > 面试题库 >

如何为多个(10k-100k)请求正确调用Akka HTTP客户端?

壤驷鸿
2023-03-14
问题内容

我正在尝试使用Akka HTTP
2.0-M2编写用于批量数据上传的工具。但是我面对akka.stream.OverflowStrategy$Fail$BufferOverflowException: Exceeded configured max-open-requests value of [32] error.

我试图找出问题,但以下示例代码也失败了:

public class TestMaxRequests {
    private static final class Router extends HttpApp {
        @Override
        public Route createRoute() {
            return route(
                    path("test").route(
                            get(handleWith(ctx -> ctx.complete("OK")))
                    )
            );
        }
    }


    public static void main(String[] args) {
        ActorSystem actorSystem = ActorSystem.create();
        Materializer materializer = ActorMaterializer.create(actorSystem);

        Router router = new Router();
        router.bindRoute("127.0.0.1", 8082, actorSystem);

        LoggingAdapter log = Logging.getLogger(actorSystem, new Object());

        for (int i = 0; i < 100; i++) {
            final int reqNum = i;
            Http.get(actorSystem).singleRequest(HttpRequest.create().withUri("http://127.0.0.1:8082/test"), materializer)
                    .onComplete(new OnComplete<HttpResponse>() {
                        @Override
                        public void onComplete(Throwable failure, HttpResponse response) throws Throwable {
                            if (failure != null) {
                                log.error(failure, "Failed: {}", reqNum);
                            } else {
                                log.info("Success: {}, consuming stream...", reqNum);
                                response.entity().getDataBytes().runWith(Sink.ignore(), materializer);
                                log.info("Success: {}, consumed stream", reqNum);
                            }
                        }
                    }, actorSystem.dispatcher());
        }
    }
}

它失败并显示:

[2015-12-15 16:17:32,609] [   INFO] [] [] a.e.s.Slf4jLogger: Slf4jLogger started
[2015-12-15 16:17:32,628] [  DEBUG] [main] [EventStream(akka://default)] a.e.EventStream: logger log1-Slf4jLogger started
[2015-12-15 16:17:32,636] [  DEBUG] [main] [EventStream(akka://default)] a.e.EventStream: Default Loggers started
[2015-12-15 16:17:33,531] [  DEBUG] [spatcher-3] [akka://default/system/IO-TCP/selectors/$a/0] a.i.TcpListener: Successfully bound to /127.0.0.1:8082
[2015-12-15 16:17:33,624] [  DEBUG] [spatcher-7] [akka://default/user/PoolInterfaceActor-0] a.h.i.e.c.PoolInterfaceActor: (Re-)starting host connection pool to 127.0.0.1:8082
[2015-12-15 16:17:33,736] [  DEBUG] [spatcher-8] [akka://default/user/SlotProcessor-0] a.h.i.e.c.PoolSlot$SlotProcessor: become unconnected, from subscriber pending
[2015-12-15 16:17:33,748] [  DEBUG] [patcher-11] [akka://default/user/SlotProcessor-3] a.h.i.e.c.PoolSlot$SlotProcessor: become unconnected, from subscriber pending
[2015-12-15 16:17:33,758] [  DEBUG] [spatcher-9] [akka://default/user/SlotProcessor-2] a.h.i.e.c.PoolSlot$SlotProcessor: become unconnected, from subscriber pending
[2015-12-15 16:17:33,762] [  DEBUG] [spatcher-9] [akka://default/user/SlotProcessor-1] a.h.i.e.c.PoolSlot$SlotProcessor: become unconnected, from subscriber pending
[2015-12-15 16:17:33,779] [  ERROR] [patcher-11] [Object(akka://default)] j.l.Object: Failed: 36
akka.stream.OverflowStrategy$Fail$BufferOverflowException: Exceeded configured max-open-requests value of [32]
    at akka.http.impl.engine.client.PoolInterfaceActor$$anonfun$receive$1.applyOrElse(PoolInterfaceActor.scala:120) ~[akka-http-core-experimental_2.11-2.0-M2.jar:na]
    at akka.actor.Actor$class.aroundReceive(Actor.scala:480) ~[akka-actor_2.11-2.4.0.jar:na]
    at akka.http.impl.engine.client.PoolInterfaceActor.akka$stream$actor$ActorSubscriber$$super$aroundReceive(PoolInterfaceActor.scala:48) ~[akka-http-core-experimental_2.11-2.0-M2.jar:na]
    at akka.stream.actor.ActorSubscriber$class.aroundReceive(ActorSubscriber.scala:201) ~[akka-stream-experimental_2.11-2.0-M2.jar:na]
    at akka.http.impl.engine.client.PoolInterfaceActor.akka$stream$actor$ActorPublisher$$super$aroundReceive(PoolInterfaceActor.scala:48) ~[akka-http-core-experimental_2.11-2.0-M2.jar:na]
    at akka.stream.actor.ActorPublisher$class.aroundReceive(ActorPublisher.scala:309) ~[akka-stream-experimental_2.11-2.0-M2.jar:na]
    at akka.http.impl.engine.client.PoolInterfaceActor.aroundReceive(PoolInterfaceActor.scala:48) ~[akka-http-core-experimental_2.11-2.0-M2.jar:na]
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:525) [akka-actor_2.11-2.4.0.jar:na]
    at akka.actor.ActorCell.invoke(ActorCell.scala:494) [akka-actor_2.11-2.4.0.jar:na]
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257) [akka-actor_2.11-2.4.0.jar:na]
    at akka.dispatch.Mailbox.run(Mailbox.scala:224) [akka-actor_2.11-2.4.0.jar:na]
    at akka.dispatch.Mailbox.exec(Mailbox.scala:234) [akka-actor_2.11-2.4.0.jar:na]
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [scala-library-2.11.7.jar:na]
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [scala-library-2.11.7.jar:na]
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [scala-library-2.11.7.jar:na]
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [scala-library-2.11.7.jar:na]
[2015-12-15 16:17:33,780] [  ERROR] [patcher-20] [Object(akka://default)] j.l.Object: Failed: 48

我猜那是因为我试图创建很多期货并立即执行它们。但是Akka不应该启用背压吗?我想我用错了。我尝试了superPool方法,但没有做任何更改,因为据我了解,Http.singleRequest内部具有相同的池。我也尝试重用Http实例而不是Http.get()在循环中调用,但这也无济于事。

触发一批请求的正确方法是什么?我计划执行1万至10万个请求的批次。


问题答案:

Akka绝对启用反压,您只是没有利用它。您可以使用单个Flow发送所有请求,而不必调度多个单个请求。从文档中:

final Flow<HttpRequest, HttpResponse, Future<OutgoingConnection>> connectionFlow = 
  Http.get(actorSystem).outgoingConnection("127.0.0.1", 8082);

然后,您可以使用此Flow处理HttpRequest对象:

HttpRequest httpRequest = HttpRequest.GET("/test")

//imitates your for-loop example of 100 requests
Source.from(() -> Collections.nCopies(100, httpRequest).iterator()) 
      .via(connectionFlow)
      .runForeach(...)


 类似资料:
  • 问题内容: 单个Servlet如何处理以用户请求形式出现的多个客户端请求?基于单例设计模式,我知道我们创建了一个servlet实例,但是单个servlet如何处理数百万个请求。对其所涉及的线程也感到困惑。 同样,这里提供了任何浏览器规范或设置,可用于跨请求发送请求或生成针对请求发送的线程。 所有框架都相同还是不同(例如,struts v / s springs)? 问题答案: Struts / S

  • 我在我的控制台上收到垃圾警告。我能找到的唯一信息是,大多数人在列表视图中启用快速滚动时会得到它。我没有使用任何列表视图。它说它被CollapsingToolbarLayout,AppCompatTextView,Toolbar,有时是AppBarLayout错误地调用。每当我快速上下“投掷”嵌套滚动视图以折叠和展开工具栏时,就会发生这种情况。 日志: xml:

  • 我希望在调试这个问题时得到一些帮助。如果我将以下JSON发送到后端,它将正常工作: 但是,如果我现在发送以下内容: 我得到了上面的错误。在我的后端代码中,我有以下代码: API 方法签名如下所示: 我认为我遇到的问题是JSON不能被解析为ZonedDateTime。有人对以下两个问题有什么建议吗?( json字符串格式ZonedDateTime自动接受什么时间,( 2)如何创建DTO来解析zone

  • 我试图找到使用httr包通过R连接Appannie的API的方法(根本没有API连接的经验)。API要求包含来自appannie网站的请求标题引用:注册App Annie帐户并生成API密钥。将此密钥添加到您的请求标头中,如下所示: 授权:持有人“”引用 我写了这样的代码 命令http_status(getdata)显示我"客户端错误:(401)未经授权"有人能帮我吗,我做错了什么?

  • 客户端的HTTP/HTTPS请求。 进程:主进程​ ClientRequest是由EventEmitter来实现Writable Stream​ new ClientRequest(options) 作用:发起新的HTTP/HTTPS请求 options(Object | String) - options是String时即请求URL。 options 是Object时则按以下属性请求: meth

  • 我目前正在玩我的球衣Rest服务。为了更好地概述给定的服务(描述、类型等),我大量使用了swagger(swagger-jersey2-jaxrs)。因此,我能够生成我的服务描述(swagger.json),我可以通过swagger ui查看和探索它们。 现在我需要创建一些客户机来使用这些服务。我遇到了swagger codegen cli,它是一个很好的工具,可以生成您的客户端和许多不同的语言(