Dòng Akka bị treo khi thực hiện các yêu cầu http qua hồ bơi kết nối

Question

Tôi đang sử dụng Akka 2.4.4 và cố di chuyển từ Apache HttpAsyncClient (không thành công).

Dưới đây là phiên bản mã đơn giản mà tôi sử dụng trong dự án của mình.

Vấn đề là nó treo nếu tôi gửi nhiều hơn 1-3 yêu cầu cho luồng. Cho đến nay sau 6 giờ gỡ lỗi tôi thậm chí không thể xác định được vấn đề. Tôi không thấy ngoại lệ, nhật ký lỗi, sự kiện trong số Decider. NOTHING :)

Tôi đã cố gắng giảm cài đặt connection-timeout xuống 1 giây khi nghĩ rằng có thể nó đang chờ phản hồi từ máy chủ nhưng nó không giúp ích gì.

Tôi đang làm gì sai?

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

Còn điều gì với số proxy support? Dường như không làm việc cho tôi.

Answer 1

Bạn cần tiêu thụ toàn bộ phần thân của phản hồi để kết nối được cung cấp cho các yêu cầu tiếp theo. Nếu bạn không quan tâm đến các đối tượng phản ứng gì cả, sau đó bạn chỉ có thể thoát nó vào một Sink.ignore, một cái gì đó như thế này:

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

Bằng cách cấu hình mặc định, khi sử dụng một hồ bơi kết nối máy chủ, các kết nối tối đa là thiết lập để 4. Mỗi hồ bơi có hàng đợi riêng của nó, nơi yêu cầu chờ đợi cho đến khi một trong các kết nối mở trở nên có sẵn. Nếu hàng đợi đó vượt quá 32 (cấu hình mặc định, có thể thay đổi, phải là lũy thừa 2) thì yo sẽ bắt đầu thấy lỗi. Trong trường hợp của bạn, bạn chỉ làm 10 yêu cầu, vì vậy bạn không đạt đến giới hạn đó. Nhưng bằng cách không tiêu thụ thực thể phản hồi, bạn không giải phóng kết nối và mọi thứ khác chỉ là hàng đợi ở phía sau, chờ kết nối miễn phí.