IT story

Akka Streams를 시작하는 방법?

hot-time 2020. 4. 22. 08:13
반응형

Akka Streams를 시작하는 방법? [닫은]

Akka Streams 라이브러리에는 이미 많은 문서가 포함되어 있습니다. 그러나 나에게 가장 큰 문제는 그것이 너무 많은 자료를 제공한다는 것입니다. 저는 배워야 할 개념의 수에 압도되어 있습니다. 거기에 나와있는 많은 예제는 매우 무겁고 실제 사용 사례로 쉽게 변환 할 수 없으므로 매우 난해합니다. 모든 빌딩 블록을 함께 구성하는 방법과 특정 문제를 해결하는 데 정확히 도움이되는 방법을 설명하지 않고 너무 많은 세부 정보를 제공한다고 생각합니다.

소스, 싱크, 흐름, 그래프 단계, 부분 그래프, 구체화, 그래프 DSL 등이 많이 있으며 어디에서 시작 해야할지 모르겠습니다. 빠른 시작 가이드는 출발점이 될하기위한 것입니다하지만 난 그것을 이해하지 않습니다. 위에서 설명한 개념을 설명하지 않고 그냥 던졌습니다. 또한 코드 예제를 실행할 수 없습니다. 텍스트를 따르기가 다소 불가능한 부분이 누락되었습니다.

누구나 개념 소스, 싱크, 흐름, 그래프 단계, 부분 그래프, 구체화 및 어쩌면 내가 놓친 다른 단어를 간단한 단어로 설명하고 모든 세부 사항을 설명하지 않는 쉬운 예제로 설명 할 수 있습니까? 시작)?

이 답변은 akka-stream버전을 기반으로 2.4.2합니다. 다른 버전에서는 API가 약간 다를 수 있습니다. 의존성은 sbt에 의해 사용될 수 있습니다 :

libraryDependencies += "com.typesafe.akka" %% "akka-stream" % "2.4.2"

좋아, 시작하자 Akka Streams의 API는 세 가지 주요 유형으로 구성됩니다. Reactive Streams 와 달리 이러한 유형은 훨씬 강력하고 복잡합니다. 모든 코드 예제에 대해 다음 정의가 이미 존재한다고 가정합니다.

import scala.concurrent._
import akka._
import akka.actor._
import akka.stream._
import akka.stream.scaladsl._
import akka.util._

implicit val system = ActorSystem("TestSystem")
implicit val materializer = ActorMaterializer()
import system.dispatcher

import문은 유형 선언 필요합니다. system는 Akka의 액터 시스템을 materializer나타내며 스트림의 평가 컨텍스트를 나타냅니다. 여기서는을 사용하는데 ActorMaterializer, 이는 액터 위에서 스트림이 평가됨을 의미합니다. 두 값 모두로 표시되므로 implicitScala 컴파일러는 필요할 때마다이 두 가지 종속성을 자동으로 주입 할 수 있습니다. 또한 system.dispatcher에 대한 실행 컨텍스트 인 import도 가져옵니다 Futures.

새로운 API

Akka Streams에는 다음과 같은 주요 속성이 있습니다.

  • 이들은 3 가지 주요 목표 역압, 비동기 및 비 차단 경계 및 서로 다른 구현 간의 상호 운용성이 Akka Streams에도 완전히 적용 되는 Reactive Streams 사양을 구현합니다 .
  • 스트림의 평가 엔진에 대한 추상화를 제공합니다 Materializer.
  • 프로그램은 세 가지 형태로 표현되고 재사용 빌딩 블록으로서 제형 Source, Sink및 Flow. 빌딩 블록은 평가를 기반으로하고 Materializer명시 적으로 트리거해야하는 그래프를 형성합니다 .

다음에는 세 가지 주요 유형을 사용하는 방법에 대한 심층적 인 소개가 제공됩니다.

출처

A Source는 데이터 작성자이며 스트림의 입력 소스 역할을합니다. 각각 Source에는 단일 출력 채널이 있고 입력 채널이 없습니다. 모든 데이터는 출력 채널을 통해에 연결된 모든 항목으로 흐릅니다 Source.

출처

boldradius.com 에서 찍은 이미지 .

A Source는 여러 가지 방법으로 만들 수 있습니다.

scala> val s = Source.empty
s: akka.stream.scaladsl.Source[Nothing,akka.NotUsed] = ...

scala> val s = Source.single("single element")
s: akka.stream.scaladsl.Source[String,akka.NotUsed] = ...

scala> val s = Source(1 to 3)
s: akka.stream.scaladsl.Source[Int,akka.NotUsed] = ...

scala> val s = Source(Future("single value from a Future"))
s: akka.stream.scaladsl.Source[String,akka.NotUsed] = ...

scala> s runForeach println
res0: scala.concurrent.Future[akka.Done] = ...
single value from a Future

위의 경우 Source에 유한 데이터를 제공하여 결국 종료됩니다. 리 액티브 스트림은 기본적으로 게으르고 비동기 적입니다. 이것은 스트림의 평가를 명시 적으로 요청해야한다는 것을 의미합니다. Akka Streams에서는 run*방법을 통해이 작업을 수행 할 수 있습니다 . 이것은 runForeach잘 알려진 foreach기능 과 다르지 않을 것 run입니다. 유한 한 데이터는 지루하기 때문에 우리는 무한한 것을 계속합니다.

scala> val s = Source.repeat(5)
s: akka.stream.scaladsl.Source[Int,akka.NotUsed] = ...

scala> s take 3 runForeach println
res1: scala.concurrent.Future[akka.Done] = ...
5
5
5

이 take방법으로 우리는 무한정의 평가를 방해하는 인공적인 정지 점을 만들 수 있습니다. 액터 지원 기능이 내장되어 있으므로 액터에게 전송되는 메시지를 스트림에 쉽게 공급할 수도 있습니다.

def run(actor: ActorRef) = {
  Future { Thread.sleep(300); actor ! 1 }
  Future { Thread.sleep(200); actor ! 2 }
  Future { Thread.sleep(100); actor ! 3 }
}
val s = Source
  .actorRef[Int](bufferSize = 0, OverflowStrategy.fail)
  .mapMaterializedValue(run)

scala> s runForeach println
res1: scala.concurrent.Future[akka.Done] = ...
3
2
1

우리는 Futures다른 스레드에서 비동기식으로 실행되는 것을 볼 수 있으며 결과를 설명합니다. 위의 예에서는 들어오는 요소에 대한 버퍼가 필요하지 않으므로 OverflowStrategy.fail버퍼 오버플로시 스트림이 실패하도록 구성 할 수 있습니다. 특히이 액터 인터페이스를 통해 모든 데이터 소스를 통해 스트림을 공급할 수 있습니다. 데이터가 동일한 스레드, 다른 스레드, 다른 프로세스 또는 인터넷을 통해 원격 시스템에서 온 것인지 여부는 중요하지 않습니다.

싱크대

A Sink는 기본적으로 a와 반대입니다 Source. 스트림의 끝점이므로 데이터를 사용합니다. A Sink에는 단일 입력 채널이 있고 출력 채널이 없습니다. Sinks스트림을 평가하지 않고 재사용 가능한 방식으로 데이터 수집기의 동작을 지정하려는 경우 특히 필요합니다. 이미 알려진 run*방법으로는 이러한 속성을 사용할 수 없으므로 Sink대신 사용하는 것이 좋습니다.

싱크대

boldradius.com 에서 찍은 이미지 .

작동의 간단한 예 Sink:

scala> val source = Source(1 to 3)
source: akka.stream.scaladsl.Source[Int,akka.NotUsed] = ...

scala> val sink = Sink.foreach[Int](elem => println(s"sink received: $elem"))
sink: akka.stream.scaladsl.Sink[Int,scala.concurrent.Future[akka.Done]] = ...

scala> val flow = source to sink
flow: akka.stream.scaladsl.RunnableGraph[akka.NotUsed] = ...

scala> flow.run()
res3: akka.NotUsed = NotUsed
sink received: 1
sink received: 2
sink received: 3

방법 Source을 사용하여에 Sink연결할 수 있습니다 to. 그것은 소위를 반환 RunnableFlow우리가 나중에 특별한 형태의 보 겠지만, 이는 Flow단지 그 호출하여 실행할 수있는 스트림 - run()방법을.

실행 가능한 흐름

boldradius.com 에서 찍은 이미지 .

싱크대에 도착한 모든 값을 액터에 전달하는 것은 물론 가능합니다 :

val actor = system.actorOf(Props(new Actor {
  override def receive = {
    case msg => println(s"actor received: $msg")
  }
}))

scala> val sink = Sink.actorRef[Int](actor, onCompleteMessage = "stream completed")
sink: akka.stream.scaladsl.Sink[Int,akka.NotUsed] = ...

scala> val runnable = Source(1 to 3) to sink
runnable: akka.stream.scaladsl.RunnableGraph[akka.NotUsed] = ...

scala> runnable.run()
res3: akka.NotUsed = NotUsed
actor received: 1
actor received: 2
actor received: 3
actor received: stream completed

흐름

Akka 스트림과 기존 시스템 사이에 연결이 필요하지만 실제로는 아무것도 할 수없는 경우 데이터 소스와 싱크가 훌륭합니다. 흐름은 Akka Streams 기본 추상화에서 마지막으로 누락 된 부분입니다. 그것들은 다른 스트림 사이의 커넥터 역할을하며 요소를 변환하는 데 사용될 수 있습니다.

흐름

boldradius.com 에서 찍은 이미지 .

a Flow가 Source새로운 것에 연결 Source되면 결과입니다. 마찬가지로에 Flow연결 Sink하면 새 가 생성 Sink됩니다. 그리고는 Flow둘 다를 연결 Source하고 SinkA의 결과 RunnableFlow. 따라서 입력 채널과 출력 채널 사이에 있지만 a Source또는 a에 연결되어 있지 않으면 맛 중 하나에 해당하지 않습니다 Sink.

풀 스트림

boldradius.com 에서 찍은 이미지 .

에 대한 이해를 돕기 위해 Flows몇 가지 예를 살펴 보겠습니다.

scala> val source = Source(1 to 3)
source: akka.stream.scaladsl.Source[Int,akka.NotUsed] = ...

scala> val sink = Sink.foreach[Int](println)
sink: akka.stream.scaladsl.Sink[Int,scala.concurrent.Future[akka.Done]] = ...

scala> val invert = Flow[Int].map(elem => elem * -1)
invert: akka.stream.scaladsl.Flow[Int,Int,akka.NotUsed] = ...

scala> val doubler = Flow[Int].map(elem => elem * 2)
doubler: akka.stream.scaladsl.Flow[Int,Int,akka.NotUsed] = ...

scala> val runnable = source via invert via doubler to sink
runnable: akka.stream.scaladsl.RunnableGraph[akka.NotUsed] = ...

scala> runnable.run()
res10: akka.NotUsed = NotUsed
-2
-4
-6

이 via방법을 통해 Source와를 연결할 수 있습니다 Flow. 컴파일러가 입력 유형을 유추 할 수 없으므로 입력 유형을 지정해야합니다. 우리는 이미 이러한 단순한 예에서 볼 수 있듯이, 흐름 invert및 double데이터 생산자와 소비자는 완전히 독립적이다. 그들은 데이터를 변환하고 출력 채널로 전달합니다. 이는 여러 스트림 사이에서 흐름을 재사용 할 수 있음을 의미합니다.

scala> val s1 = Source(1 to 3) via invert to sink
s1: akka.stream.scaladsl.RunnableGraph[akka.NotUsed] = ...

scala> val s2 = Source(-3 to -1) via invert to sink
s2: akka.stream.scaladsl.RunnableGraph[akka.NotUsed] = ...

scala> s1.run()
res10: akka.NotUsed = NotUsed
-1
-2
-3

scala> s2.run()
res11: akka.NotUsed = NotUsed
3
2
1

s1그리고 s2완전히 새로운 흐름을 나타냅니다 - 그들은 자신의 빌딩 블록을 통해 데이터를 공유하지 않습니다.

무제한 데이터 스트림

계속 진행하기 전에 먼저 Reactive Streams의 주요 측면 중 일부를 다시 방문해야합니다. 무한한 수의 요소가 어느 시점 에나 도착할 수 있으며 스트림을 다른 상태로 만들 수 있습니다. 일반적인 상태 인 실행 가능한 스트림 외에도 오류나 더 이상 데이터가 도착하지 않음을 나타내는 신호를 통해 스트림이 중지 될 수 있습니다. 다음과 같이 타임 라인에 이벤트를 표시하여 스트림을 그래픽 방식으로 모델링 할 수 있습니다.

스트림이 시간 순서대로 진행되는 일련의 진행중인 이벤트임을 나타냅니다.

누락 된 리 액티브 프로그래밍 소개 에서 가져온 이미지 .

이전 섹션의 예에서 이미 실행 가능한 흐름을 보았습니다. 우리는 얻을 RunnableGraph스트림이 실제로 구체화 할 수 있습니다 때마다이 있다고하는 수단, SinkA와 연결된다 Source. 지금까지 우리는 항상 값으로 구체화되었습니다 Unit.

val source: Source[Int, NotUsed] = Source(1 to 3)
val sink: Sink[Int, Future[Done]] = Sink.foreach[Int](println)
val flow: Flow[Int, Int, NotUsed] = Flow[Int].map(x => x)

용 Source및 Sink제 2 타입의 파라미터에 대한 Flow구체화 된 값을 나타내고 세 번째 유형 파라미터. 이 답변 전체에서 구체화의 전체 의미는 설명하지 않아야합니다. 그러나 구체화에 대한 자세한 내용은 공식 문서를 참조하십시오 . 현재 우리가 알아야 할 것은 실현 된 가치가 스트림을 실행할 때 얻는 것입니다. 우리는 지금까지 부작용에만 관심이 있었기 때문에 Unit구체화 된 가치를 얻었습니다 . 이것에 대한 예외는 싱크의 구체화로, 결과는 Future이었다. 그것은 우리에게 다시 주었다Future이 값은 싱크에 연결된 스트림이 종료 된시기를 나타낼 수 있기 때문입니다. 지금까지 이전 코드 예제는 개념을 설명하기에 좋았지 만 유한 스트림이나 매우 간단한 무한 스트림 만 다루었 기 때문에 지루했습니다. 보다 흥미롭게하기 위해, 다음에는 완전 비동기 및 무제한 스트림이 설명 될 것이다.

ClickStream 예

예를 들어 클릭 이벤트를 캡처하는 스트림이 필요합니다. 더 어렵게 만들기 위해 짧은 시간에 발생하는 클릭 이벤트를 그룹화하려고한다고 가정하겠습니다. 이렇게하면 더블, 트리플 또는 열배 클릭을 쉽게 발견 할 수 있습니다. 또한 모든 단일 클릭을 필터링하고 싶습니다. 숨을 깊이들이 쉬고 어떻게 그 문제를 반드시 해결해야하는지 상상해보십시오. 아무도 첫 번째 시도에서 올바르게 작동하는 솔루션을 구현할 수 없을 것입니다. 반응적인 방식으로이 문제를 해결하는 것은 쉽지 않습니다. 실제로이 솔루션은 구현하기가 매우 간단하고 간단하므로 코드의 동작을 직접 설명하는 다이어그램으로 솔루션을 표현할 수도 있습니다.

클릭 스트림 예의 논리

누락 된 리 액티브 프로그래밍 소개 에서 가져온 이미지 .

회색 상자는 한 스트림이 다른 스트림으로 변환되는 방법을 설명하는 함수입니다. 으로 throttle기능 우리는 250 밀리 초 이내에 클릭 축적 상기 map와 filter기능은 자명 할 것이다. 색상 오브는 이벤트를 나타내고 화살표는 이벤트를 통해 흐르는 방식을 나타냅니다. 처리 단계 후반에 우리는 스트림을 통해 흐르는 요소가 점점 줄어 듭니다. 요소를 그룹화하고 필터링하기 때문입니다. 이 이미지의 코드는 다음과 같습니다.

val multiClickStream = clickStream
    .throttle(250.millis)
    .map(clickEvents => clickEvents.length)
    .filter(numberOfClicks => numberOfClicks >= 2)

전체 논리는 단 4 줄의 코드로 표현 될 수 있습니다! 스칼라에서는 더 짧게 작성할 수 있습니다.

val multiClickStream = clickStream.throttle(250.millis).map(_.length).filter(_ >= 2)

의 정의는 clickStream조금 더 복잡하지만 예제 프로그램이 JVM에서 실행되기 때문에 발생합니다. 여기에서는 클릭 이벤트 캡처가 쉽지 않습니다. 또 다른 합병증은 Akka가 기본적으로 throttle기능을 제공하지 않는다는 것 입니다. 대신 우리는 스스로 작성해야했습니다. 이 함수는 ( map또는 filter함수 의 경우와 같이 ) 다른 사용 사례에서 재사용 할 수 있으므로이 줄을 논리 구현에 필요한 줄 수로 계산하지 않습니다. 그러나 명령형 언어에서는 논리를 쉽게 재사용 할 수없고 여러 논리 단계가 순차적으로 적용되는 대신 한 곳에서 모두 발생한다는 것이 일반적입니다. 이는 아마도 스로틀 링 논리로 코드의 모양이 잘못되었을 수 있습니다. 전체 코드 예제는요지 및 여기서 더 논의되지 않는다.

SimpleWebServer 예

대신 논의해야 할 것은 또 다른 예입니다. 클릭 스트림은 Akka Streams가 실제 예제를 처리 할 수있는 좋은 예이지만 병렬 실행을 실제로 보여줄 힘이 없습니다. 다음 예제는 여러 요청을 병렬로 처리 할 수있는 작은 웹 서버를 나타냅니다. 웹 서버는 들어오는 연결을 수락하고 인쇄 가능한 ASCII 기호를 나타내는 바이트 시퀀스를 수신 할 수 있어야합니다. 이 바이트 시퀀스 또는 문자열은 모든 개행 문자에서 더 작은 부분으로 분할되어야합니다. 그 후, 서버는 각 분할 라인으로 클라이언트에 응답해야합니다. 또는 라인으로 다른 작업을 수행하고 특별한 답변 토큰을 제공 할 수 있지만이 예제에서는 간단하게 유지하고 싶기 때문에 멋진 기능을 소개하지 않습니다. 생각해 내다, 서버는 동시에 여러 요청을 처리 할 수 ​​있어야합니다. 이는 기본적으로 다른 요청이 더 이상 실행되지 않도록 차단하는 요청이 없음을 의미합니다. Akka Streams를 사용하면 이러한 요구 사항을 모두 해결해야 할 수도 있습니다. 그러나 Akka Streams를 사용하면 이러한 요구 사항을 해결하는 데 몇 줄 이상이 필요하지 않습니다. 먼저 서버 자체에 대한 개요를 보자.

섬기는 사람

기본적으로 3 개의 주요 빌딩 블록이 있습니다. 첫 번째는 들어오는 연결을 수락해야합니다. 두 번째 요청은 들어오는 요청을 처리하고 세 번째 요청은 응답을 보내야합니다. 이 세 가지 빌딩 블록을 모두 구현하는 것은 클릭 스트림을 구현하는 것보다 조금 더 복잡합니다.

def mkServer(address: String, port: Int)(implicit system: ActorSystem, materializer: Materializer): Unit = {
  import system.dispatcher

  val connectionHandler: Sink[Tcp.IncomingConnection, Future[Unit]] =
    Sink.foreach[Tcp.IncomingConnection] { conn =>
      println(s"Incoming connection from: ${conn.remoteAddress}")
      conn.handleWith(serverLogic)
    }

  val incomingCnnections: Source[Tcp.IncomingConnection, Future[Tcp.ServerBinding]] =
    Tcp().bind(address, port)

  val binding: Future[Tcp.ServerBinding] =
    incomingCnnections.to(connectionHandler).run()

  binding onComplete {
    case Success(b) =>
      println(s"Server started, listening on: ${b.localAddress}")
    case Failure(e) =>
      println(s"Server could not be bound to $address:$port: ${e.getMessage}")
  }
}

이 함수 mkServer는 (서버의 주소와 포트를 제외하고) 액터 시스템과 materializer를 암시 적 매개 변수로 취합니다. 서버의 제어 흐름은로 표시되며 binding, 들어오는 연결의 소스를 가져 와서 들어오는 연결 싱크로 전달합니다. connectionHandler싱크 인 내부 에서는 흐름에 의한 모든 연결을 처리하며, 이에 대해서는 serverLogic나중에 설명합니다. binding를 반환Future서버가 시작되거나 시작에 실패했을 때 완료되며 다른 프로세스에서 포트를 이미 사용중인 경우에 해당됩니다. 그러나 코드는 응답을 처리하는 빌딩 블록을 볼 수 없으므로 그래픽을 완전히 반영하지 않습니다. 그 이유는 연결이 이미이 논리를 자체적으로 제공하기 때문입니다. 이전 예제에서 보았던 흐름과 같이 단방향 흐름이 아니라 양방향 흐름입니다. 구체화의 경우와 같이, 이러한 복잡한 흐름은 여기서 설명하지 않아야한다. 공식 문서는 더 복잡한 흐름 그래프를 포함하는 물질의 많음이있다. 지금은 Tcp.IncomingConnection요청 수신 방법과 응답 전송 방법을 알고있는 연결을 나타내는 것으로 충분합니다 . 여전히 누락 된 부분은serverLogic빌딩 블록. 다음과 같이 보일 수 있습니다 :

서버 로직

다시 한번, 우리는 논리를 여러 간단한 빌딩 블록으로 나눌 수 있으며, 모두 함께 프로그램의 흐름을 형성합니다. 먼저 일련의 바이트를 줄로 나누고 싶습니다. 줄 바꿈 문자를 찾을 때마다 수행해야합니다. 그 후, 원시 바이트에 대한 작업이 번거로우므로 각 행의 바이트를 문자열로 변환해야합니다. 전체적으로 복잡한 프로토콜의 이진 스트림을 수신 할 수 있으므로 들어오는 원시 데이터 작업이 매우 어려워집니다. 읽을 수있는 문자열이 있으면 답을 만들 수 있습니다. 간단한 이유로 대답은 우리의 경우에는 무엇이든 될 수 있습니다. 결국, 우리는 답을 유선을 통해 전송 될 수있는 일련의 바이트로 다시 변환해야합니다. 전체 로직의 코드는 다음과 같습니다.

val serverLogic: Flow[ByteString, ByteString, Unit] = {
  val delimiter = Framing.delimiter(
    ByteString("\n"),
    maximumFrameLength = 256,
    allowTruncation = true)

  val receiver = Flow[ByteString].map { bytes =>
    val message = bytes.utf8String
    println(s"Server received: $message")
    message
  }

  val responder = Flow[String].map { message =>
    val answer = s"Server hereby responds to message: $message\n"
    ByteString(answer)
  }

  Flow[ByteString]
    .via(delimiter)
    .via(receiver)
    .via(responder)
}

우리는 이미 알고 serverLogic걸리는 흐름입니다 ByteString과를 생산한다 ByteString. 함께 delimiter우리는 분할 할 수 있습니다 ByteString작은 부분에 - 우리의 경우는 개행 문자가 발생할 때마다 발생합니다. receiver모든 분할 바이트 시퀀스를 가져 와서 문자열로 변환하는 흐름입니다. 인쇄 가능한 ASCII 문자 만 문자열로 변환해야하지만 우리의 필요에 따라 충분하기 때문에 이것은 물론 위험한 변환입니다. responder마지막 구성 요소이며 응답을 작성하고 응답을 일련의 바이트로 다시 변환합니다. 그래픽과 달리이 마지막 구성 요소는 논리가 사소한 것이기 때문에 두 요소로 나뉘 지 않았습니다. 결국, 우리는 모든 흐름을via함수. 이 시점에서 우리는 처음에 언급 한 다중 사용자 속성을 처리했는지 묻습니다. 그리고 실제로는 분명하지 않더라도 실제로했습니다. 이 그래픽을 보면 더 명확 해집니다.

서버 및 서버 로직 결합

serverLogic성분 이외에 아무것도 작은 흐름을 포함하는 플로우 없다. 이 구성 요소는 요청 인 입력을 가져 와서 응답 인 출력을 생성합니다. 플로우는 여러 번 구성 될 수 있고 모두 독립적으로 작동하므로 다중 사용자 특성을 중첩하여 달성합니다. 모든 요청은 자체 요청 내에서 처리되므로 단기 실행 요청은 이전에 시작된 장기 실행 요청을 초과 할 수 있습니다. 궁금한 serverLogic점이 있지만 이전에 표시된 정의는 대부분 내부 정의를 인라인하여 훨씬 짧게 작성할 수 있습니다.

val serverLogic = Flow[ByteString]
  .via(Framing.delimiter(
      ByteString("\n"),
      maximumFrameLength = 256,
      allowTruncation = true))
  .map(_.utf8String)
  .map(msg => s"Server hereby responds to message: $msg\n")
  .map(ByteString(_))

웹 서버 테스트는 다음과 같습니다.

$ # Client
$ echo "Hello World\nHow are you?" | netcat 127.0.0.1 6666
Server hereby responds to message: Hello World
Server hereby responds to message: How are you?

위의 코드 예제가 제대로 작동하려면 먼저 서버를 시작해야합니다 startServer. 이 스크립트 는 스크립트로 표시됩니다.

$ # Server
$ ./startServer 127.0.0.1 6666
[DEBUG] Server started, listening on: /127.0.0.1:6666
[DEBUG] Incoming connection from: /127.0.0.1:37972
[DEBUG] Server received: Hello World
[DEBUG] Server received: How are you?

이 간단한 TCP 서버의 전체 코드 예제는 여기 에서 찾을 수 있습니다 . Akka Streams로 서버를 작성할 수있을뿐만 아니라 클라이언트도 작성할 수 있습니다. 다음과 같이 보일 수 있습니다 :

val connection = Tcp().outgoingConnection(address, port)
val flow = Flow[ByteString]
  .via(Framing.delimiter(
      ByteString("\n"),
      maximumFrameLength = 256,
      allowTruncation = true))
  .map(_.utf8String)
  .map(println)
  .map(_ ⇒ StdIn.readLine("> "))
  .map(_+"\n")
  .map(ByteString(_))

connection.join(flow).run()

전체 코드 TCP 클라이언트는 여기 에서 찾을 수 있습니다 . 코드는 매우 비슷해 보이지만 서버와 달리 더 이상 들어오는 연결을 관리 할 필요가 없습니다.

복잡한 그래프

이전 섹션에서는 흐름에서 간단한 프로그램을 구성하는 방법을 살펴 보았습니다. 그러나 실제로는 더 내장 된 함수를 사용하여 더 복잡한 스트림을 구성하는 것만으로는 충분하지 않습니다. 임의의 프로그램에 Akka Streams를 사용하려면 애플리케이션의 복잡성을 해결할 수있는 고유 한 사용자 지정 제어 구조와 결합 가능한 흐름을 작성하는 방법을 알아야합니다. 좋은 소식은 Akka Streams는 사용자의 요구에 따라 확장되도록 설계되었으며 Akka Streams의 더 복잡한 부분에 대한 짧은 소개를 제공하기 위해 클라이언트 / 서버 예제에 몇 가지 기능을 추가하는 것입니다.

아직 할 수없는 것은 연결을 닫는 것입니다. 이 시점에서 지금까지 본 스트림 API로 인해 임의의 지점에서 스트림을 중지 할 수 없기 때문에 조금 더 복잡해지기 시작합니다. 그러나 GraphStage추상화 는 몇 가지 입력 또는 출력 포트로 임의의 그래프 처리 단계를 만드는 데 사용할 수 있습니다. 먼저 서버 측을 살펴 보겠습니다. 여기서 서버라는 새로운 구성 요소를 소개합니다 closeConnection.

val closeConnection = new GraphStage[FlowShape[String, String]] {
  val in = Inlet[String]("closeConnection.in")
  val out = Outlet[String]("closeConnection.out")

  override val shape = FlowShape(in, out)

  override def createLogic(inheritedAttributes: Attributes) = new GraphStageLogic(shape) {
    setHandler(in, new InHandler {
      override def onPush() = grab(in) match {
        case "q" ⇒
          push(out, "BYE")
          completeStage()
        case msg ⇒
          push(out, s"Server hereby responds to message: $msg\n")
      }
    })
    setHandler(out, new OutHandler {
      override def onPull() = pull(in)
    })
  }
}

이 API는 흐름 API보다 훨씬 번거로워 보입니다. 당연히, 우리는 여기서 많은 명령 단계를 수행해야합니다. 그 대가로 우리는 하천의 행동을 더 잘 통제 할 수 있습니다. 위의 예에서는 하나의 입력과 하나의 출력 포트만 지정하고 shape값 을 재정 의하여 시스템에서 사용할 수 있도록합니다 . 또한 우리는 소위 InHandler및 a를 정의했는데 OutHandler,이 순서대로 요소를 수신하고 방출합니다. 전체 클릭 스트림 예제를 자세히 살펴보면 이미 이러한 구성 요소를 인식해야합니다. 에서 InHandler요소를 잡고 단일 문자가있는 문자열 인 'q'경우 스트림을 닫고 싶습니다. 고객에게 스트림이 곧 닫힐 것임을 알 수있는 기회를주기 위해 문자열을 내 보냅니다."BYE"그리고 나서 즉시 무대를 닫습니다. closeConnection성분 비아 스트림과 결합 될 수있는 via흐름에 대한 섹션에서 소개 된 방법.

연결을 닫을 수있는 것 외에도 새로 만든 연결에 환영 메시지를 표시 할 수 있다면 좋을 것입니다. 이를 위해 우리는 다시 한 번 더 나아가 야합니다.

def serverLogic
    (conn: Tcp.IncomingConnection)
    (implicit system: ActorSystem)
    : Flow[ByteString, ByteString, NotUsed]
    = Flow.fromGraph(GraphDSL.create() { implicit b ⇒
  import GraphDSL.Implicits._
  val welcome = Source.single(ByteString(s"Welcome port ${conn.remoteAddress}!\n"))
  val logic = b.add(internalLogic)
  val concat = b.add(Concat[ByteString]())
  welcome ~> concat.in(0)
  logic.outlet ~> concat.in(1)

  FlowShape(logic.in, concat.out)
})

함수는 serverLogic이제 들어오는 연결을 매개 변수로 취합니다. 본문 내에서 복잡한 스트림 동작을 설명 할 수있는 DSL을 사용합니다. 함께 welcome환영 메시지 - 우리는 하나의 요소에서 발생할 수있는 스트림을 만듭니다. logic설명 된 것입니다 serverLogic이전 섹션이다. 주목할만한 차이점은 우리가 추가했다는 closeConnection것입니다. 이제 실제로 DSL의 흥미로운 부분이 온다. 이 GraphDSL.create함수는 빌더를 b사용 가능 하게 하여 스트림을 그래프로 표현하는 데 사용됩니다. 이 ~>기능으로 입력 및 출력 포트를 서로 연결할 수 있습니다. Concat요소를 연결할 수 있습니다 예에서 사용되며, 여기에 사용되는 구성 요소가 나올 다른 요소의 앞의 환영 메시지를 앞에 추가하기internalLogic. 마지막 줄에서는 서버 논리의 입력 포트와 연결된 스트림의 출력 포트만 사용 가능하게합니다. 다른 모든 포트는 serverLogic구성 요소 의 구현 세부 사항으로 유지되어야하기 때문 입니다. Akka Streams의 DSL 그래프에 대한 자세한 소개는 공식 문서 의 해당 섹션을 참조하십시오 . 복잡한 TCP 서버와 통신 할 수있는 클라이언트의 전체 코드 예제는 여기 에서 찾을 수 있습니다 . 클라이언트에서 새 연결을 열 때마다 환영 메시지가 표시되고 "q"클라이언트 에 입력 하면 연결이 취소되었음을 알리는 메시지가 표시됩니다.

이 답변에서 다루지 않은 몇 가지 주제가 여전히 있습니다. 특히 구체화는 한 독자 나 다른 독자를 놀라게 할 수 있지만 여기서 다루는 자료는 모두가 스스로 다음 단계로 나아갈 수있을 것이라고 확신합니다. 이미 언급했듯이 공식 문서 는 Akka Streams에 대해 계속 배우기에 좋은 장소입니다.

참고 URL : https://stackoverflow.com/questions/35120082/how-to-get-started-with-akka-streams

반응형

'IT story' 카테고리의 다른 글

CSS3에서 div를 50px로 100 % 미만으로 만들 수 있습니까?  (0) 2020.04.22
오류와 경고를 파일에 기록하는 방법은 무엇입니까?  (0) 2020.04.22
유효한 UUID / GUID를 테스트하는 방법?  (0) 2020.04.22
실행중인 화면 세션에서 스크롤 백 버퍼를 늘리려면 어떻게합니까?  (0) 2020.04.22
* ngFor에 필터를 적용하는 방법?  (0) 2020.04.22

'IT story'의 다른글

  • 현재글Akka Streams를 시작하는 방법?

관련글

댓글

티스토리툴바