Как правильно вызывать HTTP-клиент Akka для нескольких (10–100 к) запросов?

Я пытаюсь написать инструмент для пакетной загрузки данных с использованием 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() в цикле, но это также не помогло.

Каков правильный способ запуска пакета запросов? Я планирую выполнять пакеты по 10 000 - 100 000 запросов.


java reactive-streams akka akka-stream akka-http
person relgames    schedule 15.12.2015    source источник

Ответы (1)


arrow_upward
18
arrow_downward

Akka абсолютно разрешает обратное давление, вы просто не пользуетесь этим. Вместо отправки нескольких отдельных запросов вы можете использовать один Flow для отправки всех ваших запросов. Из документации< /а>:

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

Затем вы можете использовать этот поток для обработки ваших HttpRequest объектов:

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

//imitates your for-loop example of 100 requests
Source.from(() -> Collections.nCopies(100, httpRequest).iterator()) 
      .via(connectionFlow)
      .runForeach(...)
person Ramón J Romero y Vigil    schedule 15.12.2015
comment
Я пропустил часть использования нескольких запросов в Source.from()! Спасибо!! - person relgames; 15.12.2015
comment
@RamonJRomeroyVigil Как я могу использовать поток с противодавлением, если я не могу заранее создать кучу запросов? Например, предположим, что я запрашиваю что-то на основе идентификаторов, возвращаемых в каком-то API с разбивкой на страницы. Таким образом, я хочу использовать поток при обработке ответов от предыдущих запросов. - person expert; 05.05.2016
comment
@expert Сочетание моего ответа выше и моего ответа на следующий вопрос должно помочь. stackoverflow.com/questions/30964824/ - person Ramón J Romero y Vigil; 06.05.2016
comment
@RamonJRomeroyVigil Спасибо, Рамон. К сожалению, у меня это не работает :( Не могли бы вы взглянуть на stackoverflow.com/q/37061585/226895 ? - person expert; 06.05.2016