final class CircuitBreaker[F[_]] extends AnyRef
The CircuitBreaker is used to provide stability and prevent
cascading failures in distributed systems.
Purpose
As an example, we have a web application interacting with a remote third party web service. Let's say the third party has oversold their capacity and their database melts down under load. Assume that the database fails in such a way that it takes a very long time to hand back an error to the third party web service. This in turn makes calls fail after a long period of time. Back to our web application, the users have noticed that their form submissions take much longer seeming to hang. Well the users do what they know to do which is use the refresh button, adding more requests to their already running requests. This eventually causes the failure of the web application due to resource exhaustion. This will affect all users, even those who are not using functionality dependent on this third party web service.
Introducing circuit breakers on the web service call would cause the requests to begin to fail-fast, letting the user know that something is wrong and that they need not refresh their request. This also confines the failure behavior to only those users that are using functionality dependent on the third party, other users are no longer affected as there is no resource exhaustion. Circuit breakers can also allow savvy developers to mark portions of the site that use the functionality unavailable, or perhaps show some cached content as appropriate while the breaker is open.
How It Works
The circuit breaker models a concurrent state machine that can be in any of these 3 states:
- Closed: During normal
operations or when the
CircuitBreakerstarts- Exceptions increment the
failurescounter - Successes reset the failure count to zero
- When the
failurescounter reaches themaxFailurescount, the breaker is tripped intoOpenstate
- Exceptions increment the
- Open: The circuit breaker
rejects all tasks with an
ExecutionRejectedException
- all tasks fail fast with
ExecutionRejectedException - after the configured
resetTimeout, the circuit breaker enters a HalfOpen state, allowing one task to go through for testing the connection
- all tasks fail fast with
- HalfOpen: The circuit breaker
has already allowed a task to go through, as a reset attempt,
in order to test the connection
- The first task when
Openhas expired is allowed through without failing fast, just before the circuit breaker is evolved into theHalfOpenstate - All tasks attempted in
HalfOpenfail-fast with an exception just as in Open state - If that task attempt succeeds, the breaker is reset back to
the
Closedstate, with theresetTimeoutand thefailurescount also reset to initial values - If the first call fails, the breaker is tripped again into
the
Openstate (theresetTimeoutis multiplied by the exponential backoff factor)
- The first task when
Usage
import monix.catnap._ import scala.concurrent.duration._ // Using cats.effect.IO for this sample, but you can use any effect // type that integrates with Cats-Effect, including monix.eval.Task: import cats.effect.{Clock, IO} implicit val clock = Clock.create[IO] // Using the "unsafe" builder for didactic purposes, but prefer // the safe "apply" builder: val circuitBreaker = CircuitBreaker[IO].unsafe( maxFailures = 5, resetTimeout = 10.seconds ) //... val problematic = IO { val nr = util.Random.nextInt() if (nr % 2 == 0) nr else throw new RuntimeException("dummy") } val task = circuitBreaker.protect(problematic)
When attempting to close the circuit breaker and resume normal operations, we can also apply an exponential backoff for repeated failed attempts, like so:
val exponential = CircuitBreaker[IO].of( maxFailures = 5, resetTimeout = 10.seconds, exponentialBackoffFactor = 2, maxResetTimeout = 10.minutes )
In this sample we attempt to reconnect after 10 seconds, then after 20, 40 and so on, a delay that keeps increasing up to a configurable maximum of 10 minutes.
Sync versus Async
The CircuitBreaker works with both
Sync and
Async
type class instances.
If the F[_] type used implements Async, then the CircuitBreaker
gains the ability to wait for it to be closed, via
awaitClose.
Retrying Tasks
Generally it's best if tasks are retried with an exponential back-off strategy for async tasks.
import cats.implicits._ import cats.effect._ import monix.execution.exceptions.ExecutionRejectedException def protectWithRetry[F[_], A](task: F[A], cb: CircuitBreaker[F], delay: FiniteDuration) (implicit F: Async[F], timer: Timer[F]): F[A] = { cb.protect(task).recoverWith { case _: ExecutionRejectedException => // Sleep, then retry timer.sleep(delay).flatMap(_ => protectWithRetry(task, cb, delay * 2)) } }
But an alternative is to wait for the precise moment at which the
CircuitBreaker is closed again and you can do so via the
awaitClose method:
def protectWithRetry2[F[_], A](task: F[A], cb: CircuitBreaker[F]) (implicit F: Async[F]): F[A] = { cb.protect(task).recoverWith { case _: ExecutionRejectedException => // Waiting for the CircuitBreaker to close, then retry cb.awaitClose.flatMap(_ => protectWithRetry2(task, cb)) } }
Be careful when doing this, plan carefully, because you might end up with the "thundering herd problem".
Credits
This Monix data type was inspired by the availability of Akka's Circuit Breaker.
- Source
- CircuitBreaker.scala
- Alphabetic
- By Inheritance
- CircuitBreaker
- AnyRef
- Any
- Hide All
- Show All
- Public
- Protected
Value Members
- final def !=(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def ##(): Int
- Definition Classes
- AnyRef → Any
- final def ==(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def asInstanceOf[T0]: T0
- Definition Classes
- Any
- def awaitClose(implicit F: OrElse[Concurrent[F], Async[F]]): F[Unit]
Awaits for this
CircuitBreakerto be closed.Awaits for this
CircuitBreakerto be closed.This only works if the type class instance used is implementing cats.effect.Async.
If this
CircuitBreakeris already in a closed state, then it returns immediately, otherwise it will wait (asynchronously) until theCircuitBreakerswitches to the Closed state again.- F
is a restriction for
F[_]to implementConcurrent[F]orAsync[F](from Cats-Effect). If it implementsConcurrent, then the resulting instance will be cancelable, to properly dispose of the registered listener in case of cancellation.
- def clone(): AnyRef
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.CloneNotSupportedException]) @native()
- def doOnClosed(callback: F[Unit]): CircuitBreaker[F]
Returns a new circuit breaker that wraps the state of the source and that will fire the given callback upon the circuit breaker transitioning to the Closed state.
Returns a new circuit breaker that wraps the state of the source and that will fire the given callback upon the circuit breaker transitioning to the Closed state.
Useful for gathering stats.
NOTE: calling this method multiple times will create a circuit breaker that will call multiple callbacks, thus the callback given is cumulative with other specified callbacks.
- callback
is to be executed when the state evolves into
Closed- returns
a new circuit breaker wrapping the state of the source
- def doOnHalfOpen(callback: F[Unit]): CircuitBreaker[F]
Returns a new circuit breaker that wraps the state of the source and that will fire the given callback upon the circuit breaker transitioning to the HalfOpen state.
Returns a new circuit breaker that wraps the state of the source and that will fire the given callback upon the circuit breaker transitioning to the HalfOpen state.
Useful for gathering stats.
NOTE: calling this method multiple times will create a circuit breaker that will call multiple callbacks, thus the callback given is cumulative with other specified callbacks.
- callback
is to be executed when the state evolves into
HalfOpen- returns
a new circuit breaker wrapping the state of the source
- def doOnOpen(callback: F[Unit]): CircuitBreaker[F]
Returns a new circuit breaker that wraps the state of the source and that will fire the given callback upon the circuit breaker transitioning to the Open state.
Returns a new circuit breaker that wraps the state of the source and that will fire the given callback upon the circuit breaker transitioning to the Open state.
Useful for gathering stats.
NOTE: calling this method multiple times will create a circuit breaker that will call multiple callbacks, thus the callback given is cumulative with other specified callbacks.
- callback
is to be executed when the state evolves into
Open- returns
a new circuit breaker wrapping the state of the source
- def doOnRejectedTask(callback: F[Unit]): CircuitBreaker[F]
Returns a new circuit breaker that wraps the state of the source and that upon a task being rejected will execute the given
callback.Returns a new circuit breaker that wraps the state of the source and that upon a task being rejected will execute the given
callback.Useful for gathering stats.
NOTE: calling this method multiple times will create a circuit breaker that will call multiple callbacks, thus the callback given is cumulative with other specified callbacks.
- callback
is to be executed when tasks get rejected
- returns
a new circuit breaker wrapping the state of the source
- final def eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- def equals(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef → Any
- val exponentialBackoffFactor: Double
A factor to use for resetting the resetTimeout when in the
HalfOpenstate, in case the attempt forClosefails. - def finalize(): Unit
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.Throwable])
- final def getClass(): Class[_ <: AnyRef]
- Definition Classes
- AnyRef → Any
- Annotations
- @native()
- def hashCode(): Int
- Definition Classes
- AnyRef → Any
- Annotations
- @native()
- final def isInstanceOf[T0]: Boolean
- Definition Classes
- Any
- val maxFailures: Int
The maximum count for allowed failures before opening the circuit breaker.
- val maxResetTimeout: Duration
The maximum timespan the circuit breaker is allowed to use as a resetTimeout when applying the exponentialBackoffFactor.
- final def ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- final def notify(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native()
- final def notifyAll(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native()
- def protect[A](task: F[A]): F[A]
Returns a new task that upon execution will execute the given task, but with the protection of this circuit breaker.
- val resetTimeout: FiniteDuration
The timespan to wait in the
Openstate before attempting a close of the circuit breaker (but without the backoff factor applied).The timespan to wait in the
Openstate before attempting a close of the circuit breaker (but without the backoff factor applied).If we have a specified exponentialBackoffFactor then the actual reset timeout applied will be this value multiplied repeatedly with that factor, a value that can be found by querying the state.
- val state: F[State]
Returns the current CircuitBreaker.State, meant for debugging purposes.
- final def synchronized[T0](arg0: => T0): T0
- Definition Classes
- AnyRef
- def toString(): String
- Definition Classes
- AnyRef → Any
- final def wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- final def wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- final def wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException]) @native()
This is the API documentation for the Monix library.
Package Overview
monix.execution exposes lower level primitives for dealing with asynchronous execution:
Atomictypes, as alternative tojava.util.concurrent.atomicmonix.catnap exposes pure abstractions built on top of the Cats-Effect type classes:
monix.eval is for dealing with evaluation of results, thus exposing Task and Coeval.
monix.reactive exposes the
Observablepattern:Observableimplementationsmonix.tail exposes Iterant for purely functional pull based streaming:
BatchandBatchCursor, the alternatives to Scala'sIterableandIteratorrespectively that we are using within Iterant's encodingYou can control evaluation with type you choose - be it Task, Coeval, cats.effect.IO or your own as long as you provide correct cats-effect or cats typeclass instance.