Come chiamare correttamente un singolo server da più attori/gestori web usando Akka HTTP?

Question

Ho un servizio (chiamiamolo Servizio A) che utilizza Akka Server HTTP per gestire le richieste in arrivo. Inoltre ho un'applicazione di terze parti (servizio B) che fornisce diversi servizi web. Lo scopo del servizio A è trasformare le richieste dei client, chiamare uno o più servizi Web del servizio B, unire/trasformare i risultati e restituirli a un cliente.

Sto utilizzando gli attori per alcune parti e solo il futuro per gli altri. Per effettuare una chiamata al servizio B, utilizzo il client HTTP Akka.

Http.get(actorSystem).singleRequest(HttpRequest.create() 
     .withUri("http://127.0.0.1:8082/test"), materializer) 
     .onComplete(...) 

Il problema è, un nuovo flusso viene creato per ogni servizio A richiesta, e se ci sono più connessioni simultanee, si traduce in akka.stream.OverflowStrategy$Fail$BufferOverflowException: Exceeded configured max-open-requests value of [32] error

ho già fatto questa domanda e ottenuto un suggerimento di utilizzare un singolo flusso How to properly call Akka HTTP client for multiple (10k - 100k) requests?

Mentre funziona per un gruppo di richieste provenienti da un singolo luogo, non so come utilizzare un singolo flusso da tutti i miei gestori simultanei di richieste.

Qual è la corretta "via Akka" per farlo?

Answer 1

Penso che si potrebbe usare Source.queue per tamponare le vostre richieste. Il codice seguente presuppone che è necessario ottenere la risposta dal servizio di terze parti, quindi avere un Future[HttpResponse] è molto gradito. In questo modo è anche possibile fornire una strategia di overflow per prevenire la fame di risorse.

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) 

(codice copiato dal mio blog post)

Answer 2

Tutto ciò che dovete fare è impostare un HostConnectionPool su Servizio B all'interno del vostro codice A di servizio. Questo ti darà un Flow che può essere aggiunto al tuo Servizio. Flussi per inviare richieste da A a B usando un pool di connessioni invece di una nuova connessione per flusso. Dalla documentazione:

A differenza di Connection livello API lato client API a livello di host si solleva da gestire manualmente i singoli connessioni HTTP. Lo gestisce autonomamente un pool configurabile di connessioni a un particolare endpoint di destinazione (cioè combinazione host/porta).

Ogni materializzazione di tale flusso, in diversi flussi, trarrà da questo pool sottostante di connessioni:

Il modo migliore per ottenere una sospensione di un pool di connessioni ad un dato bersaglio endpoint è il Metodo Http.get(system).cachedHostConnectionPool(...), che restituisce un valore Flow che può essere "cotto" in una configurazione dello stream a livello di applicazione . Questo flusso è anche chiamato "flusso del cliente del pool".

Answer 3

Ecco versione Java del accepted answer

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)); 
      } 
     }); 
}