동시성

동시성 (Concurrency)

Scala에서 병렬·동시성 애플리케이션을 작성하고 싶다면 네이티브 Java Thread를 쓸 수도 있어요. 하지만 Scala의 Future는 더 고수준이고 관용적인(idiomatic) 접근이어서 선호돼요. 그래서 이 챕터에서 Future를 다룹니다.

출처: Scala 3 Book

본문

소개

Scala Future를 그 Scaladoc에서 가져온 설명으로 먼저 보여드릴게요.

"A Future represents a value which may or may not currently be available, but will be available at some point, or an exception if that value could not be made available."

이게 무슨 뜻인지 보여드리기 위해, 먼저 단일 스레드 프로그래밍을 살펴볼게요. 단일 스레드 세계에서는 메서드 호출의 결과를 변수에 이렇게 바인딩해요.

def aShortRunningTask(): Int = 42
val x = aShortRunningTask()

이 코드에서는 값 42가 즉시 x에 바인딩돼요.

Future를 다룰 때도 할당 과정은 비슷하게 생겼어요.

def aLongRunningTask(): Future[Int] = ???
val x = aLongRunningTask()

하지만 이 경우의 핵심 차이는, aLongRunningTask가 결과를 돌려줄 때까지 걸리는 시간을 알 수 없기 때문에, x의 값이 지금은 있을 수도 없을 수도 있지만, 언젠가는—즉 미래(future)에—사용 가능해진다는 거예요.

차이를 다른 각도로 바라본다면 차단(blocking)의 관점으로 볼 수도 있어요. 이 단일 스레드 예제에서 println 문은 aShortRunningTask가 완료될 때까지 출력되지 않아요.

def aShortRunningTask(): Int =
  Thread.sleep(500)
  42
val x = aShortRunningTask()
println("Here")

반대로 aShortRunningTask가 Future로 만들어진다면, aShortRunningTask가 다른 스레드에서 실행되기 때문에 println 문이 거의 즉시 출력돼요. 차단되지 않는 거죠.

이 챕터에서는 future를 사용하는 방법을 볼게요. 여러 future를 병렬로 실행하고 그 결과를 for 표현식으로 합치는 방법도 다루죠. 그리고 future 안의 값이 반환된 뒤 그 값을 다루는 데 쓰이는 메서드들의 예제도 함께 볼게요.

future에 대해 생각할 때 꼭 알아둬야 할 점이 있어요. future는 일회성(one-shot) 구조로 설계됐다는 거예요. "이 비교적 느린 계산을 다른 스레드에서 처리하고, 끝나면 결과로 다시 불러줘" 같은 용도인 거죠. 대조적으로 Akka 액터는 오래 실행되면서 수명 동안 많은 요청에 응답하도록 설계돼 있어요. 액터는 영원히 살 수도 있지만, future는 결국 딱 한 번 실행된 계산의 결과를 담게 돼요.

REPL에서의 예제

future는 일시적인 동시성 "주머니"를 만드는 데 사용돼요. 예를 들어 실행 시간을 알 수 없는 알고리즘—원격 마이크로서비스 호출 같은—을 호출해야 해서 메인 스레드 밖에서 실행하고 싶을 때 future를 써요.

이게 어떻게 동작하는지 보여드리기 위해, REPL에서 Future 예제부터 시작할게요. 먼저 이 필수 import 문들을 붙여 넣어요.

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success}

이제 future를 만들 준비가 됐어요. 이 예제를 위해 먼저 오래 실행되는 단일 스레드 알고리즘을 정의해 볼게요.

def longRunningAlgorithm() =
  Thread.sleep(10_000)
  42

이 멋진 알고리즘은 10초의 지연 뒤 정수 값 42를 반환해요. 이제 그 알고리즘을 Future 생성자로 감싸 호출하고, 그 결과를 변수에 할당해 볼게요.

scala> val eventualInt = Future(longRunningAlgorithm())
eventualInt: scala.concurrent.Future[Int] = Future(<not completed>)

바로 그 순간, 여러분의 계산—longRunningAlgorithm() 호출—이 실행을 시작해요. eventualInt 변수의 값을 즉시 확인하면 future가 아직 완료되지 않았다는 걸 볼 수 있어요.

scala> eventualInt
val res1: scala.concurrent.Future[Int] = Future(<not completed>)

하지만 10초 뒤에 다시 확인하면 성공적으로 완료된 걸 볼 수 있어요.

scala> eventualInt
val res2: scala.concurrent.Future[Int] = Future(Success(42))

상대적으로 단순한 예제지만 기본 접근 방식을 보여줘요. 오래 걸리는 알고리즘으로 새 Future를 그냥 만들어 주면 되는 거죠.

한 가지 눈여겨볼 점은, 기대했던 42가 Success에 감싸여 있고, 그것이 또 Future에 감싸여 있다는 거예요. 이건 꼭 이해하고 넘어가야 할 핵심 개념이에요. future 안의 값은 항상 scala.util.Try 타입 중 하나, 즉 Success 또는 Failure의 인스턴스예요. 그래서 future의 결과를 다룰 때는 평소 쓰던 Try 처리 기법을 사용하면 돼요.

future에 map 사용하기

Future에는 map 메서드가 있는데, 컬렉션의 map 메서드처럼 쓰면 돼요. 변수 a를 만든 직후 map을 호출하면 이렇게 결과가 나와요.

scala> val a = Future(longRunningAlgorithm()).map(_ * 2)
a: scala.concurrent.Future[Int] = Future(<not completed>)

보시다시피 longRunningAlgorithm으로 만든 future의 초기 출력은 Future(<not completed>)예요. 하지만 10초 뒤에 a의 값을 확인하면 기대했던 결과인 84가 들어 있는 걸 볼 수 있어요.

scala> a
res1: scala.concurrent.Future[Int] = Future(Success(84))

여기서도 성공 결과가 Success와 Future 안에 감싸져 있어요.

future에 콜백 메서드 사용하기

map 같은 고차 함수 외에, future에는 콜백 메서드도 사용할 수 있어요. 흔히 쓰는 콜백 메서드 중 하나는 onComplete인데, Success와 Failure 두 경우를 처리하는 부분 함수를 인자로 받아요.

Future(longRunningAlgorithm()).onComplete {
  case Success(value) => println(s"Got the callback, value = $value")
  case Failure(e) => e.printStackTrace
}

이 코드를 REPL에 붙여 넣으면 결국 이런 결과를 보게 돼요.

Got the callback, value = 42

기타 Future 메서드

Future 클래스에는 쓸 수 있는 다른 메서드들도 있어요. Scala 컬렉션 클래스에서 볼 수 있는 메서드도 있는데, 다음을 포함해요.

• filter
• flatMap
• map

콜백 메서드는 다음과 같아요.

• onComplete
• andThen
• foreach

기타 변환 메서드로는 다음이 있어요.

• fallbackTo
• recover
• recoverWith

future에 쓸 수 있는 추가 메서드에 대한 논의는 Futures and Promises 문서를 참고하세요.

여러 future를 실행하고 결과 합치기

여러 계산을 병렬로 실행하고, 모든 future가 완료됐을 때 그 결과를 합치려면 for 표현식을 사용해요. 올바른 접근 방식은 이렇게 세 단계예요.

1. Future 결과를 반환하는 계산들을 시작한다.
2. 그 결과들을 for 표현식에서 합친다.
3. 합쳐진 결과를 onComplete나 비슷한 기법으로 추출한다.

이 세 단계가 다음 예제에 나와 있어요. 핵심은 먼저 future를 반환하는 계산들을 시작하고, 그다음에 for 표현식에서 합친다는 점이에요.

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success}

val startTime = System.currentTimeMillis()
def delta() = System.currentTimeMillis() - startTime
def sleep(millis: Long) = Thread.sleep(millis)

@main def multipleFutures1 =

  println(s"creating the futures:   ${delta()}")

  // (1) start the computations that return futures
  val f1 = Future { sleep(800); 1 }   // eventually returns 1
  val f2 = Future { sleep(200); 2 }   // eventually returns 2
  val f3 = Future { sleep(400); 3 }   // eventually returns 3

  // (2) join the futures in a `for` expression
  val result =
    for
      r1 <- f1
      r2 <- f2
      r3 <- f3
    yield
      println(s"in the 'yield': ${delta()}")
      (r1 + r2 + r3)

  // (3) process the result
  result.onComplete {
    case Success(x) =>
      println(s"in the Success case: ${delta()}")
      println(s"result = $x")
    case Failure(e) =>
      e.printStackTrace
  }

  println(s"before the 'sleep(3000)': ${delta()}")

  // important for a little parallel demo: keep the jvm alive
  sleep(3000)

이 애플리케이션을 실행하면 이런 출력이 보여요.

creating the futures:   1
before the 'sleep(3000)': 2
in the 'yield': 806
in the Success case: 806
result = 6

출력에서 볼 수 있듯이 future는 아주 빠르게 만들어져요. 메서드 끝의 sleep(3000) 문 바로 앞의 print 문이 불과 2밀리초 만에 도달하죠. 그 코드들은 모두 JVM의 메인 스레드에서 실행돼요. 그리고 806밀리초 시점에 세 future가 완료되고 yield 블록의 코드가 실행돼요. 그러면 곧바로 onComplete 메서드의 Success 케이스로 진행되죠.

806밀리초라는 출력은 세 계산이 병렬로 실행되고 있다는 걸 보여주는 열쇠예요. 순차로 실행됐다면 총 시간은 약 1,400밀리초—세 계산의 sleep 시간의 합—가 됐을 거예요. 하지만 병렬로 실행되기 때문에 총 시간은 가장 오래 걸린 계산인 f1(800밀리초)보다 아주 조금 더 길 뿐이에요.

주의할 점이 하나 있어요. 계산을 for 표현식 안에서 실행하면 순차로 실행되지, 병렬로 실행되지 않는다는 점이에요.

// Sequential execution (no parallelism!)
for
  r1 <- Future { sleep(800); 1 }
  r2 <- Future { sleep(200); 2 }
  r3 <- Future { sleep(400); 3 }
yield
  r1 + r2 + r3

그래서 계산이 병렬로 실행되길 원한다면, for 표현식 밖에서 실행하는 걸 기억하세요.

future를 반환하는 메서드

지금까지 단일 스레드 알고리즘을 Future 생성자에 넘기는 걸 봤어요. 같은 기법으로 Future를 반환하는 메서드를 만들 수도 있어요.

// simulate a slow-running method
def slowlyDouble(x: Int, delay: Long): Future[Int] = Future {
  sleep(delay)
  x * 2
}

앞선 예제들과 마찬가지로, 메서드 호출의 결과를 새 변수에 할당하면 돼요. 그러면 결과를 바로 확인했을 때 완료되지 않았다가, delay 시간이 지난 뒤에도 확인하면 future에 결과가 생긴 걸 볼 수 있어요.

scala> val f = slowlyDouble(2, 5_000L)
val f: concurrent.Future[Int] = Future(<not completed>)

scala> f
val res0: concurrent.Future[Int] = Future(<not completed>)

scala> f
val res1: concurrent.Future[Int] = Future(Success(4))

future에 관한 핵심 포인트

이 예제들로 Scala future가 어떻게 동작하는지 감이 잡혔길 바래요. 정리하자면, future에 관한 핵심 포인트 몇 가지는 이래요.

• future는 메인 스레드 밖에서 작업을 실행하기 위해 만들어요.
• future는 일회성으로, 잠재적으로 오래 실행되는 동시성 작업이 결국 값을 반환하도록 설계됐어요. 일시적인 동시성 "주머니"를 만드는 거죠.
• future는 생성하는 즉시 실행을 시작해요.
• 스레드에 비해 future가 주는 이점은 for 표현식과 잘 어울리고, 동시성 스레드를 다루는 과정을 단순화해 주는 다양한 콜백 메서드가 딸려 온다는 거예요.
• future를 다룰 때 스레드 관리의 저수준 세부사항을 신경 쓸 필요가 없어요.
• future의 결과는 onComplete와 andThen 같은 콜백 메서드, 또는 filter, map 같은 변환 메서드로 처리해요.
• Future 안의 값은 항상 Try 타입 중 하나, 즉 Success 또는 Failure의 인스턴스예요.
• 여러 future로 하나의 결과를 만들고 있다면 for 표현식에서 합쳐요.

또, 이 예제들의 import 문에서 봤듯이 Scala Future는 ExecutionContext에 의존해요.

future에 대한 더 자세한 내용은 Futures and Promises 문서를 참고하세요. 이 문서는 future, promise, execution context를 다루고, for 표현식이 flatMap 연산으로 어떻게 변환되는지에 대한 논의도 제공해요.

더 알아보기

• "Futures and Promises" 문서 (Scala 3 Book)
• Akka Actors 문서