현재 우리는 더 많은 데이터가 실시간으로 유입이 되고 있는 환경에 있습니다. 이러한 환경에서 사용되는 기술 중 하나가 스트림입니다. 스트림은 데이터를 연속적인 흐름으로 컴퓨터와 모바일 장치에 데이터를 전달하는 데 사용되는 기술입니다. 이 기술의 처리 원칙은 일부 CPU를 통해 데이터를 순차적으로 스트림으로 공급하는 것입니다. 스트림은 실시간으로 처리가 되기 때문에 IO 버퍼, Race Condition, 그리고 Latency나 throughput처럼 네트워크 관련된 문제 등도 고려해야 합니다.
Akka-Stream은 Akka-Actor를 사용해서 스트림 처리가 가능하도록 만든 기술입니다. Akka-Stream은 서로 다른 역할을 가지는 Component(여기서는 Component의 역할을 Actor 또는 Stage가 담당하도록 합니다)들의 모임이라고 할 수 있습니다. Akka-Stream의 처리는 기본적으로 아래와 같이 구성되어 있습니다
SourceAkka-Stream 처리의 시작점 입니다.
FlowAkka-Stream의 처리 과정 입니다. 하나의 입력(Source)과 하나의 출력(Sink)을 가지고 있습니다.
SinkAkka-Stream의 출구입니다.
일단 새로운 스칼라 프로젝트 생성을 위해서 sbt를 활용하기로 합니다. sbt는 스칼라에서 사용 가능한 오픈 소스 빌드 도구 입니다. 이것을 사용하면 Java의 Maven이나 Rust의 Cargo 등과 같이 필요한 패키지를 쉽게 적용할 수 있습니다. 아래와 같은 명령어를 통해서 새로 생성하도록 합니다.
sbt new scala/scala-seed.g8
프로젝트 이름까지 적용하고 나면 새로운 프로젝트가 생성됩니다. build.sbt를 통해 필요한 패키지를 추가해서 다운로드할 수 있도록 할 수 있습니다. 두 가지 패키지가 필요한데 그것은 아래와 같습니다. 아래와 같은 명령어를 추가해서 필요한 패키지를 다운로드해서 바로 적용할 수 있도록 합시다.
libraryDependencies += "com.typesafe.akka" %% "akka-stream" % "2.6.3",
libraryDependencies += "com.typesafe.akka" %% "akka-actor-typed" % "2.6.3"
일단 간단한 과정의 Akka-Stream을 만들어 보도록 하겠습니다 1부터 100까지의 입력 Source를 만들어서 그것을 출력으로 처리하는 코드입니다.
package com.example.hodong
import akka.stream._
import akka.stream.scaladsl._
import akka.{ NotUsed, Done }
import akka.actor.ActorSystem
import akka.util.ByteString
import scala.concurrent._
import scala.concurrent.duration._
object SimpleAkkaStream extends App {
implicit val system = ActorSystem("AkkaStream")
implicit val materializer = ActorMaterializer()
val source: Source[Int, NotUsed] = Source(1 to 100)
source.runForeach(i => println(i))(materializer)
}
해당 코드를 작성하고 sbt를 통해 실행시키기로 합니다. 결과는 다음과 같이 확인할 수 있습니다.
sbt "runMain com.example.hodong.SimpleAkkaStream"
미완성 현재 작성중.... Flow 작성, Sink 여러 개 작성, 입력은 TCP로 여러 개 분산 예제