कनेक्शन पूल

Question

के माध्यम से http अनुरोध करते समय अक्का फ्लो लटकता है मैं अक्का 2.4.4 का उपयोग कर रहा हूं और अपाचे HttpAsyncClient (असफल) से स्थानांतरित करने का प्रयास कर रहा हूं।

नीचे मेरे प्रोजेक्ट में उपयोग किए जाने वाले कोड का सरलीकृत संस्करण है।

समस्या यह है कि अगर मैं प्रवाह के लिए 1-3 से अधिक अनुरोध भेजता हूं तो यह लटकता है। डिबगिंग के 6 घंटों के बाद भी मैं समस्या का पता नहीं लगा सका। मुझे अपवाद, त्रुटि लॉग, Decider में ईवेंट नहीं दिख रहे हैं। कुछ भी नहीं :)

मैंने connection-timeout को 1s की सेटिंग को कम करने की कोशिश की, यह सोचकर कि यह सर्वर से प्रतिक्रिया की प्रतीक्षा कर रहा है लेकिन इससे मदद नहीं मिली।

मैं क्या गलत कर रहा हूं?

import akka.actor.ActorSystem 
import akka.http.scaladsl.Http 
import akka.http.scaladsl.model.headers.Referer 
import akka.http.scaladsl.model.{HttpRequest, HttpResponse} 
import akka.http.scaladsl.settings.ConnectionPoolSettings 
import akka.stream.Supervision.Decider 
import akka.stream.scaladsl.{Sink, Source} 
import akka.stream.{ActorAttributes, Supervision} 
import com.typesafe.config.ConfigFactory 

import scala.collection.immutable.{Seq => imSeq} 
import scala.concurrent.{Await, Future} 
import scala.concurrent.duration.Duration 
import scala.util.Try 

object Main { 

    implicit val system = ActorSystem("root") 
    implicit val executor = system.dispatcher 
    val config = ConfigFactory.load() 

    private val baseDomain = "www.google.com" 
    private val poolClientFlow = Http()(system).cachedHostConnectionPool[Any](baseDomain, 80, ConnectionPoolSettings(config)) 

    private val decider: Decider = { 
    case ex => 
     ex.printStackTrace() 
     Supervision.Stop 
    } 

    private def sendMultipleRequests[T](items: Seq[(HttpRequest, T)]): Future[Seq[(Try[HttpResponse], T)]] = 

    Source.fromIterator(() => items.toIterator) 
     .via(poolClientFlow) 
     .log("Logger")(log = myAdapter) 
     .recoverWith { 
     case ex => 
      println(ex) 
      null 
     } 
     .withAttributes(ActorAttributes.supervisionStrategy(decider)) 
     .runWith(Sink.seq) 
     .map { v => 
     println(s"Got ${v.length} responses in Flow") 
     v.asInstanceOf[Seq[(Try[HttpResponse], T)]] 
     } 

    def main(args: Array[String]) { 

    val headers = imSeq(Referer("https://www.google.com/")) 
    val reqPair = HttpRequest(uri = "/intl/en/policies/privacy").withHeaders(headers) -> "some req ID" 
    val requests = List.fill(10)(reqPair) 
    val qwe = sendMultipleRequests(requests).map { case responses => 
     println(s"Got ${responses.length} responses") 

     system.terminate() 
    } 

    Await.ready(system.whenTerminated, Duration.Inf) 
    } 
} 

इसके अलावा proxy support के साथ क्या हो रहा है? मेरे लिए काम नहीं लग रहा है।

Answer 1

आपको प्रतिक्रिया के शरीर को पूरी तरह से उपभोग करने की आवश्यकता है ताकि कनेक्शन बाद के अनुरोधों के लिए उपलब्ध कराया जा सके। आप प्रतिक्रिया संस्था के बारे में बिल्कुल भी परवाह नहीं है, तो आप बस इसे एक Sink.ignore को समाप्त हो सकती है, कुछ इस तरह:

resp.entity.dataBytes.runWith(Sink.ignore) 

डिफ़ॉल्ट config रूप से, जब एक कनेक्शन पूल मेजबान का उपयोग कर, अधिकतम कनेक्शन है 4 पर सेट करें। प्रत्येक पूल में अपनी स्वयं की कतार है जहां अनुरोध तब तक प्रतीक्षा करते हैं जब तक कि खुले कनेक्शन में से कोई भी उपलब्ध न हो जाए। यदि वह कतार कभी 32 से अधिक हो जाती है (डिफ़ॉल्ट कॉन्फ़िगरेशन, बदला जा सकता है, 2 की शक्ति होनी चाहिए) तो आप असफलताओं को देखना शुरू कर देंगे। आपके मामले में, आप केवल 10 अनुरोध करते हैं, इसलिए आप उस सीमा को नहीं दबाते हैं। लेकिन प्रतिक्रिया इकाई का उपभोग न करके आप कनेक्शन को मुक्त नहीं करते हैं और बाकी सब कुछ पीछे की कतार में हैं, कनेक्शन को मुक्त करने की प्रतीक्षा कर रहे हैं।