akka - http客户端上的断路器在失败后无法关闭

Question

一个看起来非常简单的 akka http 客户端断路器示例对我来说并不适用。

object HttpWithCircuitBreaker extends App {
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()
  implicit val ec = system.dispatcher
  val breaker =
    new CircuitBreaker(
      system.scheduler,
      maxFailures = 2,
      callTimeout = 3.seconds,
      resetTimeout = 25.seconds)
      .onOpen(println("circuit breaker opened"))
      .onClose(println("circuit breaker closed"))
      .onHalfOpen(println("circuit breaker half-open"))
  while (true) {
    val futureResponse: Future[HttpResponse] = Http().singleRequest(HttpRequest(uri = "https://www.random.org/integers/?num=1&min=1&max=6&col=1&base=10&format=plain&rnd=new"))
    breaker.withCircuitBreaker(futureResponse).map(resp => resp.status match {
      case Success(_) =>
        resp.entity.dataBytes.runWith(Sink.ignore)
        println("http success")
      case _ =>
        resp.entity.dataBytes.runWith(Sink.ignore)
        println(s"http error ${resp.status.intValue()}")
    }).recover {
      case e@_ =>
        println(s"exception ${e.getMessage}")
    }
    Thread.sleep(1000)
  }
}

它开始完美地获取随机数。它在与网络断开连接时打开，但在我重新连接时从不关闭。正如您从日志中看到的那样，尝试恢复但因超时而失败:

http success
http success
http success
exception Tcp command [Connect(www.random.org:443,None,List(),Some(10 seconds),true)] failed
exception Circuit Breaker Timed out.
circuit breaker opened
exception Tcp command [Connect(www.random.org:443,None,List(),Some(10 seconds),true)] failed
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker Timed out.
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
circuit breaker half-open             <--- new http call should start here
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker Timed out.  <--- but it fails with timeout
circuit breaker opened
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast
exception Circuit Breaker is open; calls are failing fast

直觉说它一定是一些未使用的 http 实体，但我仍然无法正确处理。

Here is the sample on github to play with

Answer 1

为了让断路器正常工作，这个val

val futureResponse: Future[HttpResponse] = Http().singleRequest(...)

应该是一个def

def futureResponse: Future[HttpResponse] = Http().singleRequest(...)

断路器需要包装异步调用 - 如果您使用 val，您的异步调用将在断路器外部启动。

此外:不要在代码中使用 Thread.sleep，尝试使用 - 例如- Akka after .