Scala 和 akka-http 休息服务的断路器答案

【问题标题】：Circuit breaker for Scala and akka-http rest serviceScala 和 akka-http 休息服务的断路器
【发布时间】：2016-03-01 13:37:50
【问题描述】：

我目前正在使用 Akka-HTTP 实现一个断路器，如下所示：

 def sendMail(entity: MyEntity): ToResponseMarshallable = {

      Thread.sleep(5 * 1000)
      validateEntity(entity).map[ToResponseMarshallable] {
        case (body, subject) if !isEmpty(body, subject) => {
          val mailResponse = sendMail(body, subject)
          OK -> ProcessedEmailMessage(mailResponse)
        }
        case _ =>
          BadRequest -> s"error: for $entity".toJson
      }
    } catch {
      case e: DeserializationException => HttpResponse(BadRequest).withEntity(HttpEntity(s"error:${e.msg}").withContentType(ContentTypes.`application/json`))
    }
  }

  val maxFailures: Int = 2
  val callTimeout: FiniteDuration = 1 second
  val resetTimeout: FiniteDuration = 30 seconds

  def open: Unit = {
    logger.info("Circuit Breaker is open")
  }

  def close: Unit = {
    logger.info("Circuit Breaker is closed")
  }

  def halfopen: Unit = {
    logger.info("Circuit Breaker is half-open, next message goes through")

  private lazy val breaker = CircuitBreaker(
    system.scheduler,
    maxFailures,
    callTimeout,
    resetTimeout
  ).onOpen(open).onClose(close).onHalfOpen(halfopen)

  def routes: Route = {
    logRequestResult("email-service_aggregator_email") {
      pathPrefix("v1") {
        path("sendmail") {
          post {
            entity(as[EmailMessage]) { entity =>
              complete {
                breaker.withCircuitBreaker(Future(sendMail(entity)))
              }
            }
          }
        }
      }
    }
  }

我的问题是，如果我使用 breaker.withCircuitBreaker(Future(sendMail(entity)))，断路器会进入打开状态，但其余响应会返回 There was an internal server error 作为响应

如果我改为使用breaker.withSyncCircuitBreaker(Future(sendMail(entity)))，那么断路器永远不会处于打开状态，但它会返回预期的HttpResponse

关于如何解决此问题以触发断路器并返回正确的 HTTP 响应的任何想法？

【问题讨论】：

你能把产生breaker的代码贴出来吗？
您需要使用onComplete 而不是complete。 complete 指令期望响应立即准备就绪。在您的情况下，withCircuitBreaker 返回Future，因此complete 将不是有效选项。 onComplete 指令设置为与Future 一起使用，因此更适合这里。然后在onComplete 回调中，您可以使用complete。
@cmbaxter 你能发个例子吗，我不明白如何用完整的方式绑定/下一步 onComplete

标签： scala akka reactive-programming akka-http

【解决方案1】：

entity(as[EmailMessage]) { entity => ctx =>
  val withBreaker = breaker.withCircuitBreaker(Future(sendMail(entity)))
  val withErrorHandling = withBreaker.recover {
      case _: CircuitBreakerOpenException => 
        HttpResponse(TooManyRequests).withEntity("Server Busy")
  }
  ctx.complete(withErrorHandling)
}

【讨论】：

它不工作，我得到同样的“有一个内部服务器错误”消息。我试图做一些调整和 sendMail() 方法直接返回一个未来，如果我不将它包装在断路器中它可以工作并且未来得到解决。
发送邮件至少需要 5 秒，而您的通话超时为 1 秒。这不会导致异常吗？
它应该进入打开状态，但没有返回响应
我已经从函数调用中删除了Thread.sleep()，并且响应正确返回。我现在想要的只是从 onOpen 函数返回一个 HttpResponse
你说得对，我创建了一个小应用程序来测试它，我的第一个请求有效，即使它需要的时间比超时时间长。接下来 30 秒内的任何后续请求都将失败。

【解决方案2】：

我将提供另一种可能的解决方案，因为我相信 onComplete 是在完成路线时处理 Future 的结果时的模式惯用方式：

entity(as[EmailMessage]) { entity =>
  val withBreaker = breaker.withCircuitBreaker(Future(sendMail(entity)))

  onComplete(withBreaker){
    case Success(trm) => 
      complete(trm)

    //Circuit breaker opened handling
    case Failure(ex:CircuitBreakerOpenException) => 
      complete(HttpResponse(TooManyRequests).withEntity("Server Busy"))

    //General exception handling
    case Failure(ex) =>
      complete(InternalServerError)
  }
}

【讨论】：