Upgrade to Pro — share decks privately, control downloads, hide ads and more …

akka-stream を始めるときに覚えておきたいこと

akka-stream を始めるときに覚えておきたいこと

akka-stream 特有の単語や概念を分かりやすく説明します。

KAWACHI Takashi

October 08, 2016
Tweet

More Decks by KAWACHI Takashi

Other Decks in Programming

Transcript

  1. akka-stream Λ࢝ΊΔͱ͖ʹ͓֮͑ͯ ͖͍ͨ͜ͱ Տ಺ ਸ Twitter: @kawachi ɹ ɹ Photo

    by Junkichi Egashira is licensed under CC BY 4.0 International 1
  2. ͸͍ɻIterator ͰͰ͖·͢ɻ def q1(fileName: String, n: Int): Boolean = {

    val src = io.Source.fromFile(fileName) val lines: Iterator[String] = src.getLines() var count = 0 while (lines.hasNext && n < count) { lines.next() count += 1 } count == n } 4
  3. ՝୊1: Ϧιʔε؅ཧͰؒҧ͍΍͍͢ def q1Again(fileName: String, n: Int): Boolean = {

    val src = io.Source.fromFile(fileName) try { val lines: Iterator[String] = src.getLines() var count = 0 while (lines.hasNext && n < count) { lines.next() count += 1 } count == n } finally { src.close() } } 5
  4. def hasMoreThan(lines: Iterator[String], n: Int) = { var count =

    0 while (lines.hasNext && count < n) { lines.next() count += 1 } count == n } def q2(fileName: String, n: Int, m: Int) = { val src = io.Source.fromFile(fileName) try { val lines: Iterator[String] = src.getLines() (hasMoreThan(lines, n), hasMoreThan(lines.filter(_.isEmpty), m)) } finally { src.close() } } 7
  5. ՝୊3: εϨου͕ block ͞ΕΔ Iterator ͸ .hasNext ͱ .next() ͕ϕʔεɻ

    while (lines.hasNext) { lines.next() // <- !!! blocking !!! … } 9
  6. • ՝୊1: Ϧιʔε؅ཧͰؒҧ͍΍͍͢ • → Ϧιʔε؅ཧΛϑΝΠϧಡΈࠐΈ Actor ʹہॴԽͰ͖Δɻͦͷ Actor ֎

    Ͱ͸ؒҧ͑ͳ͍ɻ • ՝୊2: εςʔτϑϧͳͷͰؒҧ͑΍͍͢ • → Actor Λ࡞Δࡍͷ akka.actor.Props ͸ immutable ͳ஋ɻ࠶ར༻Մೳɻ • ՝୊3: εϨου͕ block ͞ΕΔ • → ϝοηʔδۦಈͳͷͰඇಉظ API Λࣗવʹѻ͑Δ ׬શղܾʂ 11
  7. ࠓճ঺հ͢Δ akka-stream Ͱ͸ Actor ͱൺ΂ͯ • ߏ੒ཁૉ͕ग़ྗ/ೖྗ͢Δཁૉʹܕ͕͍͍ͭͯΔ • όοϑΝᷓΕ͠ͳ͍ Iterator

    ͱൺ΂ͯ • Ϧιʔε؅ཧ͕ہॴԽ͞Εؒҧ͑ʹ͍͘ • ߏ੒ཁૉ͕ immutable ͳ஋ • ඇಉظ API ͱ૊Έ߹ΘͤΒΕΔ 13
  8. ґଘੑ௥Ճ Play 2.5 ͔Β Play ͷ࣮૷ʹ akka-stream ͕࢖ΘΕ͍ͯΔ → ৽ͨʹґଘੑΛ௥Ճ͢Δඞཁͳ͠

    Play ؀ڥͰͳ͍৔߹͸ґଘੑΛ௥Ճ libraryDependencies += "com.typesafe.akka" %% "akka-stream" % "2.4.11" 15
  9. ྫ: 1~5 Λྲྀͯ͠ * 2 ͯ͠ println() import akka.actor.ActorSystem import

    akka.stream.ActorMaterializer import akka.stream.scaladsl._ val src = Source(1 to 5) val flow = Flow[Int].map(_ * 2) val sink = Sink.foreach[Int](println) val stream = src.via(flow).to(sink) implicit val system = ActorSystem() implicit val materializer = ActorMaterializer() stream.run() 17
  10. ετϦʔϜͷهड़ val src = Source(1 to 5) val flow =

    Flow[Int].map(_ * 2) val sink = Sink.foreach[Int](println) val stream = src.via(flow).to(sink) • هड़ ͱ࣮ߦ͕෼཭͞Ε͍ͯΔ • هड़͸ઃܭॻΛॻ͘࡞ۀ • ෭࡞༻ͳ͠ • src, flow, sink, stream ͷ࠶ར༻OK! 18
  11. ετϦʔϜͷ࣮ߦ • ࣮ߦ։࢝͢Δ͜ͱΛ materialize (͋Δ͍͸ run)ͱݺͿ • ࣮ߦʹ͸ Materializer ͕ඞཁ

    • Ұ౓هड़ͨ͠ετϦʔϜ͸Կ౓Ͱ΋࣮ߦͰ͖Δ implicit val system = ActorSystem() implicit val materializer = ActorMaterializer()(system) stream.run()(materializer) 22
  12. run() ͭ·Γ materialize ͷ໭Γ஋ɻ஋ ͱͯ͠ಘΒΕΔͷ͸ Materialized value ͷΈɻ ༻్ •

    ੍ޚϙΠϯτ • ֎෦͔ΒετϦʔϜΛ੍ޚ͢Δ • ؂ࢹϙΠϯτ • ࣮ߦঢ়ଶ΍ॲཧ݁Ռ 24
  13. val stream: RunnableGraph[Future[IOResult]] = FileIO.fromPath(Paths.get("src_file")) .to(FileIO.toPath(Paths.get("dest_file"))) val result: Future[IOResult] =

    stream.run() // ↑ Materialized value result.onSuccess { case ioResult => println(s"${ioResult.count} byteॲཧ͠·ͨ͠") } 25
  14. RunnableGraph ͷ(ຊ౰ͷ)ܕ RunnableGraph[+Mat] Mat ͕ materialized value ͷܕ run() ͢Δͱ

    Mat ܕͷ materialized value ͕໭Γ஋ͱͯ͠ฦ͞ ΕΔɻ 26
  15. Source, Flow, Sink ͷ(ຊ౰ͷ)ܕ Source[+Out,+Mat] Flow[-In,+Out,+Mat] Sink[-In,+Mat] ߏ੒ཁૉ (Source, Flow,

    Sink) ͕ͦΕͧΕ Mat ܕͷ materialized value Λ࣋ͭɻ ͜ΕΒΛ૊Έ߹ͤͯ RunnableGraph Λ࡞ͬͯ run() ͢Δɻ run() ͷ໭Γ஋ͱͯ͠ಘΒΕΔͷ͸ RunnableGraph ͷ Mat ͚ͩɻ 27
  16. 28

  17. 29

  18. ߹੒࣌ͷ materialized value • via(), to() ͸ࠨଆͷ materialzed value Λ߹੒ޙͷ

    materialized value ʹ࠾༻ • viaMat(), toMat() Ͱࣗ༝ʹͰ͖Δ import akka.stream.scaladsl.Keep source.viaMat(flow)(Keep.left) // ࠨͷ Mat (σϑΥϧτ) source.viaMat(flow)(Keep.right) // ӈͷ Mat source.viaMat(flow)(Keep.both) // ྆ Mat ͷλϓϧ source.viaMat(flow)(Keep.none) // NotUsed 30
  19. Source, Flow ڞ௨ͷجຊతͳૢ࡞ map(), filter(), filterNot(), take(), takeWhile(), drop(), dropWhile(),

    fold(), reduce() ඪ४ϥΠϒϥϦͷίϨΫγϣϯAPIʹࣅͨϝιου܈ Source ΍ Flow ͷग़ྗཁૉ͕ίϨΫγϣϯͷཁૉʹରԠ͢Δͱ ଊ͑Ε͹௚ײతʹૢ࡞Մೳ flatMap() ͸ͳ͍ɻmapConcat() ͋Δ͍͸ flatMapConcat() ͕૬౰ɻ 32
  20. ετϦʔϜͳΒͰ͸ͷૢ࡞ • takeWithin(d: FiniteDuration) • dropWithin(d: FiniteDuration) • ࢦఆ࣌ؒ಺ʹ౸ணͨ͠΋ͷ͚ͩऔΓग़͢/ࣺͯΔ •

    groupedWithin(n: Int, d: FiniteDuration) • ࢦఆͨ࣌ؒ͠ʹ౸ணͨ͠ཁૉʢ࠷େn݅ʣΛ·ͱΊͯ Seq Ͱग़ྗ 33
  21. Source ͷ࡞Γํ (1) • Source.apply[T](iterable: Iterable[T]): Source[T, NotUsed] Source.fromIterator[T](f: ()

    㱺 Iterator[T]): Source[T, NotUsed] • Iterable ΍ Iterator ͔Β Source Λ࡞Δ • Source.repeat[T](element: T): Source[T, NotUsed] • elementΛӬԕʹग़͢ 34
  22. Source ͷ࡞Γํ (2) • Source.tick[T](initialDelay: FiniteDuration, interval: FiniteDuration, tick: T):

    Source[T, Cancellable] • Ұఆ࣌ؒຖʹ tick Λग़͢ • Source.actorRef[T](bufferSize: Int, overflowStrategy: OverflowStrategy): Source[T, ActorRef] • materialized value ͷ ActorRef ʹϝοηʔδΛ౤͛Δͱ Source ͔ Βग़ྗ͞ΕΔ ੍ޚϙΠϯτͱͯ͠ materialized value ͕࢖ΘΕΔྫ 35
  23. 36

  24. Sink ͷ࡞Γํ • Sink.foreach[T](f: T 㱺 Unit): Sink[T, Future[Done]] •

    TܕͷཁૉΛडऔΔͨͼʹfΛ࣮ߦ͢Δ • Sink.fold[U, T](zero: U)(f: (U, T) 㱺 U): Sink[T, Future[U]] • Sink.reduce[T](f: (T, T) 㱺 T): Sink[T, Future[T]] • ৞ΈࠐΈ 37
  25. Flow ͷ࡞Γํ • Flow.apply[T]: Flow[T,T,NotUsed] • ্ྲྀ͔Βड৴ͨ͠ཁૉΛͦͷ··Լྲྀ΁ૹ৴͢Δ • Flow[Int].map(_.toString) ͷΑ͏ʹม׵ͷελʔτϙΠϯτ

    ͱͯ͠࢖͏ • Flow.fromSinkAndSource[I,O](sink: Graph[SinkShape[I],_], source: Graph[SourceShape[O],_): Flow[I,O,NotUsed] • Sink ͱ Source Λ·ͱΊͯ Flow ʹ͢Δ 38
  26. 40

  27. 41

  28. Materialize ࣌ʹɺ1 actor Ͱ࣮ߦ͢ΔΑ͏ʹࣗಈతʹ༥߹(fuse) ͞ΕΔɻ༥߹ʹ͸Φʔόϔου͕͋Δɻ͋Β͔͡Ί༥߹͢Δ͜ ͱͰ materialize ࣌ͷΦʔόʔϔουΛআڈͰ͖Δɻ val fusedStream

    = akka.stream.Fusing.aggressive(myStream) application.conf Ͱࣗಈతͳ༥߹ΛແޮԽʢσόοά༻ʣ akka.stream.materializer.auto-fusing=off 43
  29. ͱ͜ΖͰ .async ͷ࣮૷͸ override def async: Repr[Out] = addAttributes(Attributes.asyncBoundary) ଐੑ

    (attribute) ͰετϦʔϜߏ੒ཁૉΛϚʔΫɻ ࣮ߦ࣌ʹ Materializer ͕ଐੑΛݟͯ༥߹͢Δ͔൑அɻ ଞʹ͸ͲΜͳଐੑ͕ʁ 44
  30. import akka.event.Logging.WarningLevel import akka.stream.{ActorAttributes, Attributes} // ߏ੒ཁૉʹ໊લΛ෇͚Δ src.named("My source") //

    ཁૉΛϩάग़ྗ͢ΔࡍͷϩάϨϕϧ src.log("foo").withAttributes(Attributes.logLevels(WarningLevel)) // ActorMaterializer Ͱ࣮ߦ͢Δͱ͖ͷ dispatcher Λࢦఆ src.addAttributes(ActorAttributes.dispatcher("my-dispatcher")) // Τϥʔ࣌ͷ supervision strategy src.addAttributes(ActorAttributes.supervisionStrategy(decider)) 45
  31. ϦΞΫςΟϒγεςϜ(ϦΞΫςΟϒϚχϑΣετ1ΑΓ) ϝοηʔδۦಈ (Message Driven): ϦΞΫςΟϒγεςϜ͸ ඇಉ ظͳϝοηʔδύογϯά ʹґͬͯίϯϙʔωϯτؒͷڥքΛ ཱ֬͢Δɻ… ·ͨɺγεςϜ಺ʹ

    ϝοηʔδΩϡʔ Λ࡞੒ͯ͠ ؂ࢹ͠ɺඞཁͳΒ όοΫϓϨογϟʔ Λద༻͢Δ͜ͱͰϑϩʔ ੍ޚ͕ՄೳʹͳΔɻ… 1 http://www.reactivemanifesto.org/ja 47
  32. 51

  33. mapAsync(), mapAsyncUnordered() def send(email: Email): Future[X] = ??? val src:

    Source[Email] = ??? // 4 ฒྻͰϝʔϧΛૹ৴ (໭Γ஋͸ Source[X]) src.mapAsync(4)(email) // ޙ͔Βདྷͨ΋ͷͰ΋ऴΘͬͨॱʹԼྲྀ΁ྲྀ͢ src.mapAsyncUnordered(4)(email) 54
  34. Reactive streams Reactive stream ͷ PublisherɺSubscriber ͔Β Source, Sink Λ࡞Δ

    Source.fromPublisher(myPublisher) Sink.fromSubscriber(mySubscriber) 55
  35. ઐ༻ͷ stage • Reactive kafka • Akka stream contrib •

    MQTT, AMQP ͳͲ ࣗ෼Ͱ GraphStage Λ࣮૷͢Δ (akka blog) Writing Akka Streams Connectors for existing APIs 56
  36. ෳࡶͳ෼ذɺ߹ྲྀ͸ Graph DSL RunnableGraph.fromGraph(GraphDSL.create() { implicit builder => import GraphDSL.Implicits._

    val in = Source(1 to 10) val out = Sink.ignore val bcast = builder.add(Broadcast[Int](2)) val merge = builder.add(Merge[Int](2)) val f1, f2, f3, f4 = Flow[Int].map(_ + 10) in ~> f1 ~> bcast ~> f2 ~> merge ~> f3 ~> out bcast ~> f4 ~> merge ClosedShape }) 60
  37. in ~> f1 ~> bcast ~> f2 ~> merge ~>

    f3 ~> out bcast ~> f4 ~> merge ݟͨ໨͕ͦͬ͘Γʂ௚ײతʂ 61
  38. GraphDSL Ͱ͙͢࢖͑ Δ෼ذͱ߹ྲྀ • ߹ྲྀ (fan-in) • Merge, MergePreferred •

    Zip, ZipWith • ෼ذ (fan-out) • Broadcast • Balance • Unzip, UnzipWith 62
  39. ·ͱΊ • ετϦʔϜॲཧΛೃછΈͷ͋Δ API Ͱߏ੒Ͱ͖Δ • ϓϩάϥϜΛؒҧ͑ʹ͍͘Α͏ʹ޻෉͞Ε͍ͯΔ • ඇಉظ࣮ߦͱback pressure

    ʹΑΔϑϩʔ੍ޚ͕σϑΥϧτ Ͱ͍ͭͯ͘Δ • ෳࡶͳ෼ذɺ߹ྲྀ΋ graph DSL Ͱ௚ײతʹॻ͚Δ 63
  40. News: GearpumpMaterializer Apache Gearpump ͷ࣍ϦϦʔε(0.8.2)͔Β Gearpump ্Ͱ stream Λ materialize

    ͢Δ GearpumpMaterializer ؚ͕·ΕΔɻ ෼ࢄίϯϐϡʔςΟϯά + akka-stream ͷ໷໌͚ The Gearpump Materializer, Kam Kasravi 65