Skip to content

Latest commit

 

History

History
128 lines (113 loc) · 5.14 KB

README.md

File metadata and controls

128 lines (113 loc) · 5.14 KB
Raw

akka-streams

  • Akka Streams is an implementation of Reactive Streams (http://www.reactive-streams.org/)
  • Reactive Streams is a Specification of asynchronous stream processing with non-blocking pressure
  • Reactive Streams API Consists of the following components that are required to by It's implementations
    • Publisher --> Emits Unbounded elements (asynchronously)
    • Subscriber --> Receives elements
    • Subscription
    • Processor --> Transforms elements along the way
    • async
    • back-pressure

Akka-Streams (Implementation of Reactive Streams)

  • Source ~ Publisher : Emits elements asynchronously, may or may not terminate
  • Sink ~ Subscriber : Receives elements, terminates only when the publisher terminates
  • Flow ~ Processor : Transforms Elements
  • Build Streams by connection components (Operators)
  • Directions
    • upstream -> towards the source - - - - -> Source
    • Downstream -> towards the sink <- - - - -
  • Graph : Upstream --> Source --> Flow --> Flow --> Sink

Materializing Value's

  • Getting a meaningful value out of a running Stream
  • Components are static until they run
 val graph = source.via(flow).to(sink)
 val result = graph.run() ==> result -> Materialized value

  • A graph is a blueprint for a stream

  • Running a graph allocates the right resources

    • Creating actor's
    • Allocation thread pools, sockets, connections, etc... everything is transparent

  • Running a graph is Materializing

  • Materializing a graph = materializing all components in that graph

  • Each component produces a materialized value when run

  • the graph produces a single materialized value (by default leftmost component value Keep.left)

    • can change by using xxxMat(Keep.xxx()), Keep contains #left, #right, #both (left, right), #none
    • our job to choose which one to pick

  • A component can materialize multiple times

    • you can reuse the same component in different graphs
    • different runs = different materialization's!

  • A materialized value can be ANYTHING

    • NotUsed -> combination of scala's Unit & Java's Void
    • Future[Any]
    • etc...

Operator Fusion And Async Boundaries

  • Akka Stream Components fused by default => The entire graph will run on the same actor
  • To make sure each component to run on a separate actor introduce Async boundaries
  private val simpleSource: Source[Int, NotUsed] = Source(1 to 15)
  private val adder: Flow[Int, Int, NotUsed] = Flow[Int].map { element =>
    println(s"[${Thread.currentThread().getName}] Adder for Element : $element")
    element + 1
  }
  private val multiplier: Flow[Int, Int, NotUsed] = Flow[Int].map{ element =>
    println(s"[${Thread.currentThread().getName}] Adder for Element : $element")
    element * 10
  }
  private val sink: Sink[Int, Future[Done]] = Sink.foreach[Int](element => println(s"[${Thread.currentThread().getName}] Printing Element : $element"))

  // This runs on the SAME ACTOR => This is called as operator/component FUSION
  private val used: NotUsed = simpleSource
    .via(adder)
    .via(multiplier)
    .to(sink)
    .run()
  • The above stream will turn into below model when we materialize(call runXXX()) the static graph(stream)
class SimpleActor extends Actor {
    override def receive: Receive = {
      case x : Int =>
        // Flows
        println(s"[${Thread.currentThread().getName}] Adder for Element : $x")
        val x2 = x + 1
        println(s"[${Thread.currentThread().getName}] Adder for Element : $x2")
        val y = x2 * 10
        // Sink
        println(s"[${Thread.currentThread().getName}] Printing Element : $y")

    }
  }
  private val simpleActor: ActorRef = actorSystem.actorOf(Props[SimpleActor])
  (1 to 15).foreach(simpleActor ! _)

Async boundaries

source.async
      .via(Flow[Int].map(x => x + 1)).async
      .via(Flow[Int].map(x => x * 1)).async
      .to(Sink.foreach(println))

* Components run on different actors
* better throughput
  • Best when : Individual operations are expensive
  • Avoid when : Operations are comparable with a message pass
  • Irrespective of Async or not, streams always guarantees Ordering
 simpleSource
    .via(complexAdder).async // From Here Runs on one actor => [OperatorFusionAndAsyncBoundaries-akka.actor.default-dispatcher-8] Complex Adder for Element : 1
    .via(complexMultiplier).async // From Here Runs on another Actor => [OperatorFusionAndAsyncBoundaries-akka.actor.default-dispatcher-5] Complex Multiplier for Element : 2
    .to(sink) // From Here Runs on Another Actor => [OperatorFusionAndAsyncBoundaries-akka.actor.default-dispatcher-6] Printing Element : 20
    .run()
  • An async boundary contains
    • Everything from the previous boundary (if any)
    • Everything between the previous boundary and current boundary
  • Communication based on actor messages

BackPressure

  • One of the fundamental features of Reactive Streams
  • Elements flow as response to demand from consumers
  • Fast Consumers : all is well
  • Slow Consumers: Problem
    • consumer will send a signal to producer to slow down using Backpressure Protocol
  • Akka Streams can slow down fast producers
  • Backpressure protocol is transparent

Streams Graph