如何正确调用Akka HTTP客户端多个（10k - 100k）请求？ [英] How to properly call Akka HTTP client for multiple (10k - 100k) requests?
问题描述
我正在尝试使用Akka HTTP 2.0-M2编写批量数据上传工具。
但是我面临 akka.stream.OverflowStrategy $ Fail $ BufferOverflowException：超出配置的最大开放请求值[32]错误。
I'm trying to write a tool for batch data upload using Akka HTTP 2.0-M2.
But I'm facing akka.stream.OverflowStrategy$Fail$BufferOverflowException: Exceeded configured max-open-requests value of [32] error.
我试图找出一个问题，这里的示例代码也失败了：
I tried to isolate a problem and here is the sample code which also fails:
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
我想这是因为我正在尝试创建大量期货并立即执行它们。但阿卡不应该启用背压吗？我想我错了。我尝试过superPool方法但没有改变，因为据我所知， Http.singleRequest 里面有相同的池。我也尝试重用Http实例，而不是在循环中调用 Http.get（），但它也没有帮助。
I guess that's because I'm trying to create a lot of Futures and execute them all at once. But isn't Akka supposed to enable backpressure? I guess I'm using it wrong. I tried superPool methods but nothing changed because, as I understand, Http.singleRequest has same pool inside. I also tried to reuse Http instance instead of calling Http.get() in the loop but it also did not help.
解雇一批请求的正确方法是什么？我打算执行10 000 - 10万个批次的请求。
What is the correct way of firing a batch of requests? I am planning to execute batches of 10 000 - 100 000 requests.
推荐答案
Akka绝对能够实现背压，你只是不服用它的优点。您可以使用单个 Flow 来发送所有请求，而不是分派多个单个请求。来自文档：
Akka absolutely enables backpressure, you're just not taking advantage of it. Instead of dispatching multiple single requests, you can use a single Flow to send all of your requests through. From the documentation:
final Flow<HttpRequest, HttpResponse, Future<OutgoingConnection>> connectionFlow =
Http.get(actorSystem).outgoingConnection("127.0.0.1", 8082);
然后您可以使用此流程来处理您的 HttpRequest 对象：
You can then use this Flow to process your HttpRequest objects:
HttpRequest httpRequest = HttpRequest.GET("/test")
//imitates your for-loop example of 100 requests
Source.from(() -> Collections.nCopies(100, httpRequest).iterator())
.via(connectionFlow)
.runForeach(...)
这篇关于如何正确调用Akka HTTP客户端多个（10k - 100k）请求？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
