如何正确调用Akka HTTP客户端多个（10k – 100k）请求？
我正在尝试使用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() { @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] [] [] aesSlf4jLogger: Slf4jLogger started [2015-12-15 16:17:32,628] [ DEBUG] [main] [EventStream(akka://default)] aeEventStream: logger log1-Slf4jLogger started [2015-12-15 16:17:32,636] [ DEBUG] [main] [EventStream(akka://default)] aeEventStream: Default Loggers started [2015-12-15 16:17:33,531] [ DEBUG] [spatcher-3] [akka://default/system/IO-TCP/selectors/$a/0] aiTcpListener: Successfully bound to /127.0.0.1:8082 [2015-12-15 16:17:33,624] [ DEBUG] [spatcher-7] [akka://default/user/PoolInterfaceActor-0] ahiecPoolInterfaceActor: (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] ahiecPoolSlot$SlotProcessor: become unconnected, from subscriber pending [2015-12-15 16:17:33,748] [ DEBUG] [patcher-11] [akka://default/user/SlotProcessor-3] ahiecPoolSlot$SlotProcessor: become unconnected, from subscriber pending [2015-12-15 16:17:33,758] [ DEBUG] [spatcher-9] [akka://default/user/SlotProcessor-2] ahiecPoolSlot$SlotProcessor: become unconnected, from subscriber pending [2015-12-15 16:17:33,762] [ DEBUG] [spatcher-9] [akka://default/user/SlotProcessor-1] ahiecPoolSlot$SlotProcessor: become unconnected, from subscriber pending [2015-12-15 16:17:33,779] [ ERROR] [patcher-11] [Object(akka://default)] jlObject: 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)] jlObject: Failed: 48
我想这是因为我正在尝试创建大量期货并立即执行所有期货。 但阿卡不应该启用背压吗？ 我想我错了。 我尝试过superPool方法但没有改变，因为据我所知， Http.singleRequest内部有相同的池。 我也尝试重用Http实例而不是在循环中调用Http.get() ，但它也没有帮助。
解雇一批请求的正确方法是什么？ 我计划执行10 000到10万个批次的批量请求。
Akka绝对能够实现背压，你只是没有利用它。 您可以使用单个Flow来发送所有请求，而不是分派多个单个请求。 从文档 ：
final Flow> 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(...)