複数の（10k〜100k）リクエストに対してAkka HTTPクライアントを正しく呼び出す方法は？

Question

私は、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.get()を呼び出す代わりにHttpインスタンスを再利用しようとしましたが、それはまた助けになりませんでした。

リクエストのバッチを実行する正しい方法は何ですか？私は10 000 - 100 000リクエストのバッチを実行する予定です。

Answer 1

Akkaは絶対にバックプレッシャーを可能にします。あなたはそれを利用していません。複数の単一リクエストをディスパッチする代わりに、単一のFlowを使用してすべてのリクエストを送信することができます。 documentationから：

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

次に、あなたのHttpRequestオブジェクトを処理するために、このフローを使用することができます。

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

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