我有一个服务(我们称之为服务A),它使用Akka Server HTTP来处理传入的请求。我还有提供多种 Web 服务的第三方应用程序(服务 B)。服务 A 的目的是转换客户端请求,调用服务 B 的一个或多个 Web 服务,合并/转换结果并将其提供回客户端。
我在某些部分使用Actor,而在其他部分则只使用未来。为了调用服务 B,我使用 Akka HTTP 客户端。
Http.get(actorSystem).singleRequest(HttpRequest.create()
.withUri("http://127.0.0.1:8082/test"), materializer)
.onComplete(...)
问题是,每个服务 A 请求都会创建一个新流,如果有多个并发连接,则会导致akka.stream.OverflowStrategy$Fail$BufferOverflowException: Exceeded configured max-open-requests value of [32] error
我已经问过这个问题,并得到了使用单个流的建议 如何为多个(10k - 100k)请求正确调用 Akka HTTP 客户端?
虽然它适用于来自单个位置的一批请求,但我不知道如何使用所有并发请求处理程序中的单个 Flow。
正确的"阿卡方式"是什么?
我认为您可以使用Source.queue
来缓冲您的请求。下面的代码假设您需要从第三方服务中获取答案,因此非常欢迎Future[HttpResponse]
。这样,您还可以提供溢出策略来防止资源匮乏。
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.stream.scaladsl.{Keep, Sink, Source}
import akka.stream.{ActorMaterializer, OverflowStrategy}
import scala.concurrent.duration._
import scala.concurrent.{Await, Future, Promise}
import scala.util.{Failure, Success}
import scala.concurrent.ExecutionContext.Implicits.global
implicit val system = ActorSystem("main")
implicit val materializer = ActorMaterializer()
val pool = Http().cachedHostConnectionPool[Promise[HttpResponse]](host = "google.com", port = 80)
val queue = Source.queue[(HttpRequest, Promise[HttpResponse])](10, OverflowStrategy.dropNew)
.via(pool)
.toMat(Sink.foreach({
case ((Success(resp), p)) => p.success(resp)
case ((Failure(e), p)) => p.failure(e)
}))(Keep.left)
.run
val promise = Promise[HttpResponse]
val request = HttpRequest(uri = "/") -> promise
val response = queue.offer(request).flatMap(buffered => {
if (buffered) promise.future
else Future.failed(new RuntimeException())
})
Await.ready(response, 3 seconds)
(代码复制自我的博客文章)
这是 Java 版本的接受答案
final Flow<
Pair<HttpRequest, Promise<HttpResponse>>,
Pair<Try<HttpResponse>, Promise<HttpResponse>>,
NotUsed> flow =
Http.get(actorSystem).superPool(materializer);
final SourceQueue<Pair<HttpRequest, Promise<HttpResponse>>> queue = Source.<Pair<HttpRequest, Promise<HttpResponse>>>
queue(BUFFER_SIZE, OverflowStrategy.dropNew())
.via(flow)
.toMat(Sink.foreach(p -> p.second().complete(p.first())), Keep.left())
.run(materializer);
...
public CompletionStage<HttpResponse> request(HttpRequest request) {
log.debug("Making request {}", request);
Promise<HttpResponse> promise = Futures.promise();
return queue.offer(Pair.create(request, promise))
.thenCompose(buffered -> {
if (buffered instanceof QueueOfferResult.Enqueued$) {
return FutureConverters.toJava(promise.future())
.thenApply(resp -> {
if (log.isDebugEnabled()) {
log.debug("Got response {} {}", resp.status(), resp.getHeaders());
}
return resp;
});
} else {
log.error("Could not buffer request {}", request);
return CompletableFuture.completedFuture(HttpResponse.create().withStatus(StatusCodes.SERVICE_UNAVAILABLE));
}
});
}
您需要做的就是在服务 A 代码中设置一个到 Service B 的 HostConnectionPool。 这将为您提供一个可以添加到服务 A 流中的Flow
,以使用连接池(而不是每个流的新连接)将请求从 A 调度到 B。 从文档中:
与连接级客户端 API 相反,主机级 API 无需手动管理单个 HTTP 连接。它 自主管理可配置的连接池 特定目标端点(即主机/端口组合)。
此流的每个具体化(在不同的流中)都将从此基础连接池中提取:
将连接池保留到给定目标的最佳方法 端点是
Http.get(system).cachedHostConnectionPool(...)
方法, 它返回一个可以"烘焙"到应用程序级别的Flow
流设置。此流也称为"池客户端流"。