Post revision changes:
- Moved sample code in source files - Text review - Proper use of RST syntax - Improved ScalaDoc
This commit is contained in:
galarragas
parent
457ccc1250
commit
fd5b872a43
8 changed files with 210 additions and 187 deletions
|
|
@ -1,184 +1,38 @@
|
|||
Circuit-Breaker Actor
|
||||
======================
|
||||
.. _circuit-breaker:
|
||||
|
||||
This is an alternative implementation of the [AKKA Circuit Breaker Pattern](http://doc.akka.io/docs/akka/snapshot/common/circuitbreaker.html).
|
||||
The main difference is that is intended to be used only for request-reply interactions with actor using the Circuit-Breaker as a proxy of the target one
|
||||
in order to provide the same failfast functionalities and a protocol similar to the AKKA Pattern implementation
|
||||
Circuit-Breaker Actor
|
||||
=====================
|
||||
|
||||
This is an alternative implementation of the [Akka Circuit Breaker Pattern](http://doc.akka.io/docs/akka/snapshot/common/circuitbreaker.html).
|
||||
The main difference is that it is intended to be used only for request-reply interactions with an actor using the Circuit-Breaker as a proxy of the target one
|
||||
in order to provide the same failfast functionalities and a protocol similar to the circuit-breaker implementation in Akka.
|
||||
|
||||
|
||||
### Usage
|
||||
|
||||
Let's assume we have an actor wrapping a back-end service and able to respond to `Request` calls with a `Response` object
|
||||
containing an `Either[String, String]` to map successful and failed responses. The service is also potentially slowing down
|
||||
Let's assume we have an actor wrapping a back-end service and able to respond to ``Request`` calls with a ``Response`` object
|
||||
containing an ``Either[String, String]`` to map successful and failed responses. The service is also potentially slowing down
|
||||
because of the workload.
|
||||
|
||||
A simple implementation can be given by this class::
|
||||
A simple implementation can be given by this class
|
||||
|
||||
|
||||
object SimpleService {
|
||||
case class Request(content: String)
|
||||
case class Response(content: Either[String, String])
|
||||
case object ResetCount
|
||||
}
|
||||
|
||||
/**
|
||||
* This is a simple actor simulating a service
|
||||
* - Becoming slower with the increase of frequency of input requests
|
||||
* - Failing around 30% of the requests
|
||||
*/
|
||||
class SimpleService extends Actor with ActorLogging {
|
||||
import SimpleService._
|
||||
|
||||
var messageCount = 0
|
||||
|
||||
import context.dispatcher
|
||||
|
||||
context.system.scheduler.schedule(1.second, 1.second, self, ResetCount)
|
||||
|
||||
override def receive = {
|
||||
case ResetCount =>
|
||||
messageCount = 0
|
||||
|
||||
case Request(content) =>
|
||||
messageCount += 1
|
||||
// simulate workload
|
||||
Thread.sleep( 100 * messageCount )
|
||||
// Fails around 30% of the times
|
||||
if(Random.nextInt(100) < 70 ) {
|
||||
sender ! Response(Right(s"Successfully processed $content"))
|
||||
} else {
|
||||
sender ! Response(Left(s"Failure processing $content"))
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
.. includecode:: @contribSrc@/src/test/scala/akka/contrib/circuitbreaker/sample/SimpleService.scala#simple-service
|
||||
|
||||
|
||||
If we want to interface with this service using the Circuit Breaker we can use two approaches:
|
||||
|
||||
Using a non-conversational approach: ::
|
||||
Using a non-conversational approach:
|
||||
|
||||
class CircuitBreakerExample(potentiallyFailingService: ActorRef) extends Actor with ActorLogging {
|
||||
import SimpleService._
|
||||
.. includecode:: @contribSrc@/src/test/scala/akka/contrib/circuitbreaker/sample/CircuitBreaker.scala#basic-sample
|
||||
|
||||
val serviceCircuitBreaker =
|
||||
context.actorOf(
|
||||
CircuitBreakerActorBuilder( maxFailures = 3, callTimeout = 2.seconds, resetTimeout = 30.seconds )
|
||||
.copy(
|
||||
failureDetector = {
|
||||
_ match {
|
||||
case Response(Left(_)) => true
|
||||
case _ => false
|
||||
}
|
||||
}
|
||||
)
|
||||
.propsForTarget(potentiallyFailingService),
|
||||
"serviceCircuitBreaker"
|
||||
)
|
||||
Using the ``ask`` pattern, in this case it is useful to be able to map circuit open failures to the same type of failures
|
||||
returned by the service (a ``Left[String]`` in our case):
|
||||
|
||||
override def receive: Receive = {
|
||||
case AskFor(requestToForward) =>
|
||||
serviceCircuitBreaker ! Request(requestToForward)
|
||||
|
||||
case Right(Response(content)) =>
|
||||
//handle response
|
||||
log.info("Got successful response {}", content)
|
||||
|
||||
case Response(Right(content)) =>
|
||||
//handle response
|
||||
log.info("Got successful response {}", content)
|
||||
|
||||
case Response(Left(content)) =>
|
||||
//handle response
|
||||
log.info("Got failed response {}", content)
|
||||
|
||||
case CircuitOpenFailure(failedMsg) =>
|
||||
log.warning("Unable to send message {}", failedMsg)
|
||||
}
|
||||
}
|
||||
|
||||
Using the ASK pattern, in this case it is useful to be able to map circuit open failures to the same type of failures
|
||||
returned by the service (a `Left[String]` in our case): ::
|
||||
|
||||
class CircuitBreakerAskExample(potentiallyFailingService: ActorRef) extends Actor with ActorLogging {
|
||||
import SimpleService._
|
||||
import akka.pattern._
|
||||
|
||||
implicit val askTimeout: Timeout = 2.seconds
|
||||
|
||||
val serviceCircuitBreaker =
|
||||
context.actorOf(
|
||||
CircuitBreakerActorBuilder( maxFailures = 3, callTimeout = askTimeout, resetTimeout = 30.seconds )
|
||||
.copy(
|
||||
failureDetector = {
|
||||
_ match {
|
||||
case Response(Left(_)) => true
|
||||
case _ => false
|
||||
}
|
||||
}
|
||||
)
|
||||
.copy(
|
||||
openCircuitFailureConverter = { failure =>
|
||||
Left(s"Circuit open when processing ${failure.failedMsg}")
|
||||
}
|
||||
)
|
||||
.propsForTarget(potentiallyFailingService),
|
||||
"serviceCircuitBreaker"
|
||||
)
|
||||
|
||||
import context.dispatcher
|
||||
|
||||
override def receive: Receive = {
|
||||
case AskFor(requestToForward) =>
|
||||
(serviceCircuitBreaker ? Request(requestToForward)).mapTo[Either[String, String]].onComplete {
|
||||
case Success(Right(successResponse)) =>
|
||||
//handle response
|
||||
log.info("Got successful response {}", successResponse)
|
||||
|
||||
case Success(Left(failureResponse)) =>
|
||||
//handle response
|
||||
log.info("Got successful response {}", failureResponse)
|
||||
|
||||
case Failure(exception) =>
|
||||
//handle response
|
||||
log.info("Got successful response {}", exception)
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
.. includecode:: @contribSrc@/src/test/scala/akka/contrib/circuitbreaker/sample/CircuitBreakerAsk.scala#ask-sample
|
||||
|
||||
|
||||
If it is not possible to define define a specific error response, you can map the Open Circuit notification into a failure.
|
||||
That also means that your `CircuitBreakerActor` will be essentially useful to protect you from time out for extra workload or
|
||||
temporary failures in the target actor ::
|
||||
If it is not possible to define define a specific error response, you can map the Open Circuit notification to a failure.
|
||||
That also means that your ``CircuitBreakerActor`` will be useful to protect you from time out for extra workload or
|
||||
temporary failures in the target actor includecode:: code/docs/stream/io/StreamFileDocSpec.scala#file-source
|
||||
|
||||
class CircuitBreakerAskWithFailureExample(potentiallyFailingService: ActorRef) extends Actor with ActorLogging {
|
||||
import SimpleService._
|
||||
import akka.pattern._
|
||||
import CircuitBreakerActor._
|
||||
|
||||
implicit val askTimeout: Timeout = 2.seconds
|
||||
|
||||
val serviceCircuitBreaker =
|
||||
context.actorOf(
|
||||
CircuitBreakerActorBuilder( maxFailures = 3, callTimeout = askTimeout, resetTimeout = 30.seconds ).propsForTarget(potentiallyFailingService),
|
||||
"serviceCircuitBreaker"
|
||||
)
|
||||
|
||||
import context.dispatcher
|
||||
|
||||
override def receive: Receive = {
|
||||
case AskFor(requestToForward) =>
|
||||
(serviceCircuitBreaker ? Request(requestToForward)).failForOpenCircuit.mapTo[String].onComplete {
|
||||
case Success(successResponse) =>
|
||||
//handle response
|
||||
log.info("Got successful response {}", successResponse)
|
||||
|
||||
case Failure(exception) =>
|
||||
//handle response
|
||||
log.info("Got successful response {}", exception)
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
.. includecode:: @contribSrc@/src/test/scala/akka/contrib/circuitbreaker/sample/CircuitBreakerAskWithFailure.scala#ask-with-failure-sample
|
||||
|
|
|
|||
|
|
@ -30,7 +30,6 @@ object CircuitBreakerActor {
|
|||
* response from a request represent a failure
|
||||
* @param failureMap function to map a failure into a response message. The failing response message is wrapped
|
||||
* into a [[akka.contrib.circuitbreaker.CircuitBreakerActor.CircuitOpenFailure]] object
|
||||
* @return
|
||||
*/
|
||||
def props(target: ActorRef,
|
||||
maxFailures: Int,
|
||||
|
|
@ -60,21 +59,6 @@ object CircuitBreakerActor {
|
|||
|
||||
final case class CircuitBreakerStateData(failureCount: Int = 0, firstHalfOpenMessageSent: Boolean = false)
|
||||
|
||||
/**
|
||||
* Convenience builder with appropriate defaults to create the CircuitBreakerActor props
|
||||
*
|
||||
* @param maxFailures maximum number of failures before opening the circuit
|
||||
* @param callTimeout timeout before considering the ongoing call a failure
|
||||
* @param resetTimeout time after which the channel will be closed after entering the open state
|
||||
* @param circuitEventListener an actor that will receive a series of messages of type
|
||||
* [[akka.contrib.circuitbreaker.CircuitBreakerActor.CircuitBreakerEvent]]
|
||||
* defaults to None
|
||||
* @param failureDetector function to detect if the a message received from the target actor as
|
||||
* response from a request represent a failure, defaults to a function accepting every
|
||||
* response making this circuit breaker be activated by response timeouts only
|
||||
* @param openCircuitFailureConverter function to map a failure into a response message.
|
||||
* Defaults to an identify function
|
||||
*/
|
||||
final case class CircuitBreakerActorBuilder(
|
||||
maxFailures: Int, callTimeout: Timeout, resetTimeout: Timeout,
|
||||
circuitEventListener: Option[ActorRef] = None,
|
||||
|
|
@ -85,7 +69,6 @@ object CircuitBreakerActor {
|
|||
* Creates the props for a [[akka.contrib.circuitbreaker.CircuitBreakerActor]] proxying the given target
|
||||
*
|
||||
* @param target the target actor ref
|
||||
* @return
|
||||
*/
|
||||
def propsForTarget(target: ActorRef) = CircuitBreakerActor.props(target, maxFailures, callTimeout, resetTimeout, circuitEventListener, failureDetector, openCircuitFailureConverter)
|
||||
|
||||
|
|
@ -97,8 +80,6 @@ object CircuitBreakerActor {
|
|||
* Extends [[scala.concurrent.Future]] with the method failForOpenCircuitWith to handle
|
||||
* [[akka.contrib.circuitbreaker.CircuitBreakerActor.CircuitOpenFailure]] failure responses throwing
|
||||
* an exception built with the given exception builder
|
||||
*
|
||||
* @param future
|
||||
*/
|
||||
implicit class CircuitBreakerAwareFuture(val future: Future[Any]) extends AnyVal {
|
||||
def failForOpenCircuit(implicit executionContext: ExecutionContext): Future[Any] = failForOpenCircuitWith(new OpenCircuitException)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
/**
|
||||
* Copyright (C) 2014-2015 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
* Copyright (C) 2014-2015 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
package akka.contrib.circuitbreaker
|
||||
|
||||
import akka.actor.ActorRef
|
||||
|
|
|
|||
|
|
@ -0,0 +1,46 @@
|
|||
package akka.contrib.circuitbreaker.sample
|
||||
|
||||
import akka.actor.{ Actor, ActorLogging, ActorRef }
|
||||
import akka.contrib.circuitbreaker.CircuitBreakerActor.{ CircuitBreakerActorBuilder, CircuitOpenFailure }
|
||||
|
||||
import scala.concurrent.duration._
|
||||
|
||||
//#basic-sample
|
||||
class CircuitBreaker(potentiallyFailingService: ActorRef) extends Actor with ActorLogging {
|
||||
import SimpleService._
|
||||
|
||||
val serviceCircuitBreaker =
|
||||
context.actorOf(
|
||||
CircuitBreakerActorBuilder(maxFailures = 3, callTimeout = 2.seconds, resetTimeout = 30.seconds)
|
||||
.copy(
|
||||
failureDetector = {
|
||||
_ match {
|
||||
case Response(Left(_)) ⇒ true
|
||||
case _ ⇒ false
|
||||
}
|
||||
})
|
||||
.propsForTarget(potentiallyFailingService),
|
||||
"serviceCircuitBreaker")
|
||||
|
||||
override def receive: Receive = {
|
||||
case AskFor(requestToForward) ⇒
|
||||
serviceCircuitBreaker ! Request(requestToForward)
|
||||
|
||||
case Right(Response(content)) ⇒
|
||||
//handle response
|
||||
log.info("Got successful response {}", content)
|
||||
|
||||
case Response(Right(content)) ⇒
|
||||
//handle response
|
||||
log.info("Got successful response {}", content)
|
||||
|
||||
case Response(Left(content)) ⇒
|
||||
//handle response
|
||||
log.info("Got failed response {}", content)
|
||||
|
||||
case CircuitOpenFailure(failedMsg) ⇒
|
||||
log.warning("Unable to send message {}", failedMsg)
|
||||
}
|
||||
}
|
||||
//#basic-sample
|
||||
|
||||
|
|
@ -0,0 +1,54 @@
|
|||
package akka.contrib.circuitbreaker.sample
|
||||
|
||||
import akka.actor.{ Actor, ActorLogging, ActorRef }
|
||||
import akka.contrib.circuitbreaker.CircuitBreakerActor.CircuitBreakerActorBuilder
|
||||
import akka.util.Timeout
|
||||
|
||||
import scala.concurrent.duration._
|
||||
import scala.util.{ Failure, Success }
|
||||
|
||||
//#ask-sample
|
||||
class CircuitBreakerAsk(potentiallyFailingService: ActorRef) extends Actor with ActorLogging {
|
||||
import SimpleService._
|
||||
import akka.pattern._
|
||||
|
||||
implicit val askTimeout: Timeout = 2.seconds
|
||||
|
||||
val serviceCircuitBreaker =
|
||||
context.actorOf(
|
||||
CircuitBreakerActorBuilder(maxFailures = 3, callTimeout = askTimeout, resetTimeout = 30.seconds)
|
||||
.copy(
|
||||
failureDetector = {
|
||||
_ match {
|
||||
case Response(Left(_)) ⇒ true
|
||||
case _ ⇒ false
|
||||
}
|
||||
})
|
||||
.copy(
|
||||
openCircuitFailureConverter = { failure ⇒
|
||||
Left(s"Circuit open when processing ${failure.failedMsg}")
|
||||
})
|
||||
.propsForTarget(potentiallyFailingService),
|
||||
"serviceCircuitBreaker")
|
||||
|
||||
import context.dispatcher
|
||||
|
||||
override def receive: Receive = {
|
||||
case AskFor(requestToForward) ⇒
|
||||
(serviceCircuitBreaker ? Request(requestToForward)).mapTo[Either[String, String]].onComplete {
|
||||
case Success(Right(successResponse)) ⇒
|
||||
//handle response
|
||||
log.info("Got successful response {}", successResponse)
|
||||
|
||||
case Success(Left(failureResponse)) ⇒
|
||||
//handle response
|
||||
log.info("Got successful response {}", failureResponse)
|
||||
|
||||
case Failure(exception) ⇒
|
||||
//handle response
|
||||
log.info("Got successful response {}", exception)
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
//#ask-sample
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
package akka.contrib.circuitbreaker.sample
|
||||
|
||||
import akka.actor.{ Actor, ActorLogging, ActorRef }
|
||||
import akka.contrib.circuitbreaker.CircuitBreakerActor._
|
||||
import akka.util.Timeout
|
||||
|
||||
import scala.concurrent.duration._
|
||||
import scala.util.{ Failure, Success }
|
||||
|
||||
//#ask-with-failure-sample
|
||||
class CircuitBreakerAskWithFailure(potentiallyFailingService: ActorRef) extends Actor with ActorLogging {
|
||||
import SimpleService._
|
||||
import akka.pattern._
|
||||
|
||||
implicit val askTimeout: Timeout = 2.seconds
|
||||
|
||||
val serviceCircuitBreaker =
|
||||
context.actorOf(
|
||||
CircuitBreakerActorBuilder(maxFailures = 3, callTimeout = askTimeout, resetTimeout = 30.seconds).propsForTarget(potentiallyFailingService),
|
||||
"serviceCircuitBreaker")
|
||||
|
||||
import context.dispatcher
|
||||
|
||||
override def receive: Receive = {
|
||||
case AskFor(requestToForward) ⇒
|
||||
(serviceCircuitBreaker ? Request(requestToForward)).failForOpenCircuit.mapTo[String].onComplete {
|
||||
case Success(successResponse) ⇒
|
||||
//handle response
|
||||
log.info("Got successful response {}", successResponse)
|
||||
|
||||
case Failure(exception) ⇒
|
||||
//handle response
|
||||
log.info("Got successful response {}", exception)
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
//#ask-with-failure-sample
|
||||
|
|
@ -0,0 +1,45 @@
|
|||
package akka.contrib.circuitbreaker.sample
|
||||
|
||||
import akka.actor.{ ActorLogging, Actor }
|
||||
import scala.concurrent.duration._
|
||||
import scala.util.Random
|
||||
|
||||
//#simple-service
|
||||
object SimpleService {
|
||||
case class Request(content: String)
|
||||
case class Response(content: Either[String, String])
|
||||
case object ResetCount
|
||||
}
|
||||
|
||||
/**
|
||||
* This is a simple actor simulating a service
|
||||
* - Becoming slower with the increase of frequency of input requests
|
||||
* - Failing around 30% of the requests
|
||||
*/
|
||||
class SimpleService extends Actor with ActorLogging {
|
||||
import SimpleService._
|
||||
|
||||
var messageCount = 0
|
||||
|
||||
import context.dispatcher
|
||||
|
||||
context.system.scheduler.schedule(1.second, 1.second, self, ResetCount)
|
||||
|
||||
override def receive = {
|
||||
case ResetCount ⇒
|
||||
messageCount = 0
|
||||
|
||||
case Request(content) ⇒
|
||||
messageCount += 1
|
||||
// simulate workload
|
||||
Thread.sleep(100 * messageCount)
|
||||
// Fails around 30% of the times
|
||||
if (Random.nextInt(100) < 70) {
|
||||
sender ! Response(Right(s"Successfully processed $content"))
|
||||
} else {
|
||||
sender ! Response(Left(s"Failure processing $content"))
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
//#simple-service
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
package akka.contrib.circuitbreaker
|
||||
|
||||
package object sample {
|
||||
case class AskFor(what: String)
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue