1. Future
2. Promise
3. Awiat
4. async
5. Observable
6. 병렬
Future
스칼라에서의 Future 는 꽤 다양한 방식으로 사용 할 수있는데
먼저 스칼라에서의 Future 모양을 살펴보자.
trait Future[T]
: 퓨쳐값을 나타낸다. T 타입의 실제 우리가 리턴받기 원하는 객체를 포함한다.
def apply[T](b: =>T) (implicit e: ExecutionContext) : Future[T]
: 퓨처 계산을 나타낸다. 실제 계산을 수행하는 함수를 매개변수로 넣어주고 있다.
: 암시적으로 ExecutionContext 가 매개변수로 들어간다. 즉 쓰레드풀을 넣어주는것.
퓨쳐값 과 퓨쳐계산을 잘 구분해서 기억해두자.
1) 퓨쳐 실행
object FuturesComputation extends App {
import scala.concurrent._
import ExecutionContext.Implicits.global
Future {
log(s"the future is here")
}
log(s"the future is coming")
}2개의 로그중 어떤것이 먼저 출력되는지는 비 결정적이다. (물론 아래 log 가 먼저 찍힐 가능성이 크겠지만) Future 를 통해서 자연스럽게 첫번째 로그는 자신의 쓰레드안에서 독립적으로 실행된다.
Future{...} 는 예상했다시피 Future.apply 메소드 호출을 생략되진 것이다. 스칼라에서 함수호출시 () 대신 {} 를 이용할 수 있다. 단 인자가 하나일 경우에만. {..} 의 해당내용은 apply 의 매개변수로 들어 갈 것이며 Future 값을 반환 할 것이다.
2) 퓨쳐 실행 그리고 Future 리턴
object FuturesDataType extends App {
import scala.concurrent._
import ExecutionContext.Implicits.global
import scala.io.Source
val myFuture: Future[String] = Future {
val f = Source.fromFile("build.sbt")
try f.getLines.mkString("\n") finally f.close()
}
log(s"started reading build file asynchronously")
log(s"status: ${myFuture.isCompleted}")
Thread.sleep(250)
log(s"status: ${myFuture.isCompleted}")
log(s"status: ${myFuture.value}")
}
- 파일을 읽는 일을 Future 계산을 통해서 위임하고 바로 Future [String] 값을 리턴받는다.
- 리턴 받은 Future 에서 계산 완료가 되었는지 계속 확인 한 후에 (즉 계속 폴링) 실제 결과 값을 찍어준다.
- 이렇게 Future 를 사용하면 사실 비동기를 사용하는 의미가 줄어 든다.
예를들어 스타벅스에서 1분마다 카운터에 가서 계속 내 커피가 나왔는지 물어 보지 말고, 커피가 나왔으면 커피를 내 자리로 바로 배달해주도록 하는 것도 커피를 주문 할 때 알려주면 안될까?
자 그 역할을 하는 콜백함수를 Future 에 넣어보자.
3) 퓨쳐 와 콜백
object FuturesCallbacks extends App {
import scala.concurrent._
import ExecutionContext.Implicits.global
import scala.io.Source
def getUrlSpec(): Future[Seq[String]] = Future {
val f = Source.fromURL("http://www.w3.org/Addressing/URL/url-spec.txt")
try f.getLines.toList finally f.close()
}
val urlSpec: Future[Seq[String]] = getUrlSpec()
def find(lines: Seq[String], word: String) = lines.zipWithIndex collect {
case (line, n) if line.contains(word) => (n, line)
} mkString("\n")
urlSpec foreach {
lines => log(s"Found occurrences of 'telnet'\n${find(lines, "telnet")}\n")
}
urlSpec foreach {
lines => log(s"Found occurrences of 'password'\n${find(lines, "password")}\n")
}
log("callbacks installed, continuing with other work")
Thread.sleep(1000 * 10)
}
- 웹페이지를 긁어 오는 Future 를 실행하고 퓨처값[리스트]을 리턴 받는다.
- foreach 메소드를 통해서 최종완료된 값을 처리해 준다. (콜백처리)
- 가져온 데이터에서 "telnet" , "password" 가 있는 라인만 추려서 출력해준다.
3) 퓨쳐 와 예외
object FuturesFailure extends App {
import scala.concurrent._
import ExecutionContext.Implicits.global
import scala.io.Source
val urlSpec: Future[String] = Future {
Source.fromURL("http://www.w3.org/non-existent-url-spec.txt").mkString
//Source.fromURL("http://www.w3.org/Addressing/URL/url-spec.txt").mkString
}
// urlSpec foreach {
// value => log(s"Found occurrences of 'telnet'\n${value}\n")
// }
urlSpec.failed foreach {
case t => log(s"exception occurred - $t")
}
Thread.sleep(1000*10)
}
- 보통 Future 계산에서 예외를 던지면 그에 대응하는 Future 값은 값으로 완료 될 수 없다.
- 스칼라 Future 는 성공적으로 완료되거나, 실패로 완료된다. 퓨처가 실패로 완료되는 경우를 퓨쳐가 실패 했다라고 말한다.
- foreach 메소드는 성공적으로 완료한 퓨처에서만 값을 받는 콜백만을 받는다.
- 따라서 실패 콜백을 지정하기 위해서는 다른 메소드가 필요하다. 그게 failed 라 한다.
- failed 메소드는 퓨처가 실패하면서 내놓는 예외를 포함하는 Future[Throwable] 객체를 반환한다.
object FuturesExceptions extends App {
import scala.concurrent._
import ExecutionContext.Implicits.global
import scala.io.Source
import scala.util.{Try, Success, Failure}
val file = Future { Source.fromFile(".gitignore-SAMPLE").getLines.mkString("\n") }
file foreach {
text => log(text)
}
file.failed foreach {
case fnfe: java.io.FileNotFoundException => log(s"Cannot find file - $fnfe")
case t => log(s"Failed due to $t")
}
file onComplete {
case Success(text) => log(text)
case Failure(t) => log(s"Failed due to $t")
}
Thread.sleep(1000*10)
}
- 간결하게 onComplete 를 통해서도 성공과 실패를 처리 함께 할 수 있다.
- onComplete 함수안에는 Try[T] 가 패턴매칭으로 들어간다.
object FuturesTry extends App {
import scala.util._
val threadName: Try[String] = Try(Thread.currentThread.getName)
val someText: Try[String] = Try("Try objects are created synchronously")
val message: Try[String] = for {
tn <- threadName
st <- someText
} yield s"$st, t = $tn"
message match {
case Success(msg) => log(msg)
case Failure(error) => log(s"There should be no $error here.")
}
Thread.sleep(1000*10)
}
- Try[T] 타입은 Option[T] 와 비슷하지만 다른점은 실패시 정보를 추가 할 수 있다는 점이다. Option[T] 같은 경우 그냥 None 일 뿐이지 않는가.
- Try[T] 는 Future[T] 와 달리 동기적으로 다루어진다.
4) 퓨쳐 와 합성
map
콜백은 유용하지만 프로그램이 커지면 이를 사용한 프로그램 흐름에 대한 추론이 어려워 진다.
또한 콜백을 사용하면 특정 패턴의 비동기 프로그래밍을 사용할 수 없게 된다. 특히 여러 퓨처에 한꺼번에 한 콜백을 등록하는 것은 귀찮은 일이 된다.
val buildFile = Future { Source.fromFile("build.sbt").getLines }
val longestBuildLine = buildFile.map(lines => lines.maxBy(_.length))
longestBuildLine onComplete {
case Success(line) => log(s"the longest line is '$line'")
}
map 은 기존 퓨처에 있는 값을 사용해서 새 퓨처를 만드는 메소드이다.
def map[S](f: T => S) (implicit e : ExecutionContext) : Future[S]
위의 예에서는 1. build.sbt 파일에서 각 라인들을 읽어서 2. 그 중에서 가장 긴 라인을 출력하는 2가지의 함수가 합성되었다.이걸 쓰레드를 직접만들어서 했을 때와 비교하면 얼마나 간단해 졌는지 실감 할 수 있을 것이다.
for complehension
val gitignoreFile = Future { Source.fromFile(".gitignore-SAMPLE").getLines }
val longestGitignoreLine = for (lines <- gitignoreFile) yield lines.maxBy(_.length)
longestBuildLine onComplete {
case Success(line) => log(s"the longest line is '$line'")
}다음처럼 for comprehension 으로 처리 할 수 도 있다.
스칼라에서는 map 메소드가 있는 객체에 대해서는 for comprehension 을 허용한다.
async - await
아래 처럼 Future 를 async-await 구문을 사용하여, 명령형 언어처럼 순차적으로 쓸 수 있어서 가독성이 증가 된다.
import scala.concurrent.ExecutionContext.Implicits.global
import scala.async.Async._
val gitignoreFile1: Future[Int] = ...
val gitignoreFile2: Future[Int] = ...
val asyncComputation = async {
await(gitignoreFile1) + await(gitignoreFile2)
}
flatMap
먼저 스칼라 List 에서의 map 과 flatMap의 차이를 살펴보자.
scala> val fruits = Seq("apple", "banana", "orange")
fruits: Seq[java.lang.String] = List(apple, banana, orange)
scala> fruits.map(_.toUpperCase)
res0: Seq[java.lang.String] = List(APPLE, BANANA, ORANGE)
scala> fruits.flatMap(_.toUpperCase)
res1: Seq[Char] = List(A, P, P, L, E, B, A, N, A, N, A, O, R, A, N, G, E)map 은 받아드린 리스트의 하나의 값에 대해서 따로 실행하여 개별 List 를 만들었고
flatMap 은 따로 실행하여 큰 뭉치의 List 로 만들었다. 즉 하나의 큰 List 로 평평하게 펴서 나열했다는 의미.
또 다른 flatMap 의 특징을 알아보자.
def toInt(s: String): Option[Int] = {
try {
Some(Integer.parseInt(s.trim))
} catch {
// catch Exception to catch null 's'
case e: Exception => None
}
}
scala> val strings = Seq("1", "2", "foo", "3", "bar")
strings: Seq[java.lang.String] = List(1, 2, foo, 3, bar)
scala> strings.map(toInt)
res0: Seq[Option[Int]] = List(Some(1), Some(2), None, Some(3), None)
scala> strings.flatMap(toInt)
res1: Seq[Int] = List(1, 2, 3)
scala> strings.flatMap(toInt).sum
res2: Int = 6 None 은 무시하고 Some 타입만, 타입을 벗겨서 처리하고있다.
이제 Future 에서 어떻게 flatMap을 정의하였는지 살펴보자.
val netiquette = Future { Source.fromURL("http://www.ietf.org/rfc/rfc1855.txt").mkString }
val urlSpec = Future { Source.fromURL("http://www.w3.org/Addressing/URL/url-spec.txt").mkString }
val answer = netiquette.flatMap { nettext =>
urlSpec.map { urltext =>
"First, read this: " + nettext + ". Now, try this: " + urltext
}
}
answer foreach {
case contents => log(contents)
}FlatMap 을 사용하였다. 이것은 Map과는 다르게 현재 퓨처가 완료되고 그 결과값에 f 를 적용해서 생기는 새로운 퓨처가 완료되어야만 완료된다. 즉 아래와 같다.
def map[S](f: T => S) (implicit e : ExecutionContext) : Future[S]
def flatMap[S](f: T => Future[S]) (implicit e : ExecutionContext) : Future[S]
위의 코드는 결국 nettext 라는 값을 A퓨처의 결과로받고, urltext 라는 값 또한 B라는 퓨처의 결과로 받아서 하나의 처리를 거친 후에 answer 라는 새로운 Future[S] 를 반환한다는것을 말해주고있다.
object FuturesFlatMap extends App {
import scala.concurrent._
import ExecutionContext.Implicits.global
import scala.io.Source
val netiquette = Future { Source.fromURL("http://www.ietf.org/rfc/rfc1855.txt").mkString }
val urlSpec = Future { Source.fromURL("http://www.w3.org/Addressing/URL/url-spec.txt").mkString }
val answer = for {
nettext <- netiquette
urltext <- urlSpec
} yield {
"First of all, read this: " + nettext + " Once you're done, try this: " + urltext
}
answer foreach {
case contents => log(contents)
}
Thread.sleep(1000*10)
}보기 간단하게 for comprehesion 으로 바꾸었다.
for comprehension 보다 더 이해하기 쉬운 방법으로는 async, awit 가 있는데 여기서는 생략한다.
다음과 같은 Data Flow 로 그려볼 수 있다. Reactive 파라다임에서 말하는
"모든 지점에서 블럭 되지 않게 하자. 자연스럽게 이벤트가 흘러다니도록 하자." 이 되겠다.
5) 이후에 알아야 할 것들
기존 콜백기반의API와 퓨처사이에 다리를 놓고 잘 조합하기 위해서는 퓨처계산을 원하는 대로 정의할수있는 Promise를 알아야한다. 또한 퓨처들의 종료를 기다렸다가 무언가를 처리해야 하는 경우에 (블록이 가끔 필요하다) Await.ready 와 Await.result 등도 알아야한다.
위의 모든것들을 편하게 사용하게 하는 스칼라 비동기 라이브러리가 있다. 바로 async 와 await 인데 이것을 사용하면 조금은 더 편하게 Future 를 활용할 수 있다.
마지막으로 결과를 한번만 처리 할 수 있는 퓨처말고 동일한 계산으로 부터 여러 번 서로 다른 이벤트를 받아야 하는 경우 ( 파일 다운로드 상태 추적등) 에는 이벤트 스트림에 대해서 알아야한다. 이러한 이벤트 스트림을 Observable[T] 라는 타입으로 표현했는데 이에 관한 내용도 알아야한다. Akka Streaming 이라는 도구도 있다.
6) 한꺼번에 이해하기
아래 카카오에서 발행한 글을 통해 다시 복습해서 이해해보면 좋을 거 같다.
카카오톡 - 스칼라 Future 를 통한 모나드 이해
7) 비동기식 퓨처 연산 정리
설명 |
예제 |
설명 |
failbackTo |
nextFtr(1) failbackTo nextFtr(2) |
두번째 퓨쳐를 첫번째 연결하고 새로운 종합적인 퓨처를 반환함. 첫 번째 퓨처가 성공적이지 않다면 두 번째 퓨처가 호출됨 |
flatMap |
nextFtr(1).flatMap(int => nextFtr()) |
두번째 퓨처를 첫 번째에 연결하고 새로운 종합적인 퓨처를 반환함. 첫 번째가 성공적이라면 그 반환 값이 두 번째를 호출하는데 사용됨 |
map |
nextFtr(1) map (_ * 2) |
주어진 함수를 퓨처에 연결하고 새로운 종합적인 퓨처를 반환함. 퓨처가 성공적이라면 그 반환 값이 해당 함수를 호출할 때 사용됨 |
onComplete |
nextFtr() onComplete
{ _ getOrElse 0 } |
퓨처의 작업이 완료된 후 주어진 함수가 값 또는 예외를 포함한 Try를 이용하여 호출됨 |
onFailure |
* 2.12 부터 사장됨 |
|
onSuccess |
* 2.12 부터 사장됨 |
|
Future.sequence |
concurrent.Future sequence List(nextFtre(1),nextFtr(5)) |
주어진 시퀀스에서 퓨처를 벙행으로 실행하여 새로운 퓨처를 반환함. 시퀀스 내의 모든 퓨처가 성공하면 이들의 반환값의 리스트가 반환됨. 그렇지 않으면 그 시퀀스내에서 처음으로 발생한 예외가 반환됨 |
* 러닝스칼라 (제이슨스와츠,제이펍 발생) 에서 발췌
레퍼런스:
1.스칼라 동시성 프로그래밍
2.러닝 스칼라