KotlinConf 2019 : Roman Elizarov의 Kotlin Flow를 사용한 비동기 데이터 스트림 강연 정리

[ChanJun-Park]

Recap on kotlin coroutines

Asynchronous yet sequential

코틀린에서는 buildList 라는 api 를 통해 여러 값의 리스트를 suspend 함수들의 호출을 통해 생성하여 반환할 수 있다. 그러나 이 코드는 모든 값이 계산되고 나서야 결과로 받은 리스트를 통해 작업을 수행할 수 있는 문제가 있다.

만약 작업 할 수 있는 요소가 생성될때마다 그 값을 받아서 처리하고 싶다면 Channel 을 이용할 수 있다. Channel 은 두 Coroutine 간에 데이터를 주고 받을 수 있는 파이프라인이다.

이 Channel 을 통해서 작업 할 수 있는 요소가 생성될때 마다 채널에 데이터를 전송해서 수신자 측에서 그 값을 처리할 수 있도록 개선할 수 있다.

그러나 Channel 을 사용한 방법에도 문제가 있다. Channel 은 hot 하다. 즉, 수신자가 없더라도 요소들을 Channel 에 전송할 수 있다.

관례적으로 Channel 을 생성하는 함수를 CoroutineScope 의 확장함수 형태로 만들어 이 채널이 수신자와 상관없이 독립적으로 실행될 수 있음을 나타낼 수 있지만, 여전히 에러 발생에 취약한 코드를 작성할 위험이 있다.

이런 문제를 해결하고자 등장한 것이 Kotlin Flow 이다.

Kotlin Flow

Flow 는 flow 라는 빌더 함수를 통해 생성할 수 있다. Flow 에서 데이터를 전달하겠다는 명령은 emit 함수를 통해 수행할 수 있다. 이전의 Channel 과는 다르게 flow 빌더를 통해 Flow 를 생성했다 하더라도 별도의 coroutine 이 생성되어 곧바로 값들을 emit 하는 것이 아니라 누군가 Flow 를 collect 했을때만 값이 계산되기 시작한다.

Flow is cold

앞서 언급했지만 Flow 는 cold 이다. 누군가가 collect 하기 전까지는 어떠한 작업도 수행하지 않기 때문에 앞서 Channel 이 가지고 있던 문제를 해결할 수 있다.

Flow is declarative

또다른 Flow 의 속성으로 Flow 는 Declarative 한 프로그래밍 스타일을 만든다는 것이다. 아래 코드는 Flow 가 collect 될때 어떤 작업을 수행할 것인지를 선언만한다. 실제로 값이 계산되어 emit 되는 시점은 collect 함수와 같은 terminal operator 를 실행한 시점이다.

fun foo(): Flow<String> = flow {
  emit("A")
  emit("B")
  emit("C")
}

flowOf() 라는 함수를 통해 간단하게 Flow 를 생성해줄 수도 있는데, list builder 와 비교해보자.

listOf() 함수를 통해 여러 값의 리스트를 만들때는 코드가 곧바로 실행되기 때문에 Run - Imperative 한 스타일의 코드가 된다. 또한 list 빌더 내부에서 곧바로 suspend 함수를 호출하기 때문에 foo 함수 자체에도 suspend 를 붙여줘야 한다.

반면 flowOf() 함수를 통해 Flow 를 만들때에는 코드가 곧바로 실행되는 것이 아니라 이 Flow 가 collect 되었을때 어떤 값들이 계산될지를 선언만 해주기 때문에 Defined - Declarative 한 스타일의 코드가 된다. 또한 코드가 곧바로 실행되는 것이 아니기 때문에 foo 함수 자체에 suspend 를 붙여주지 않아도 된다. 실제로 Flow 내부 코드를 실행하기 시작하는 collect 와 같은 terminal operator 함수에는 suspend 가 붙어있어 어느 시점에 Flow 내부 코드가 실행될지를 쉽게 파악할 수 있다.

실행 방식에도 차이가 있는데 list builder 는 모든 리스트의 값을 한번에 계산하는 반면, Flow 는 하나의 값이 emit 될때마다 reactive 하게 하나씩 계산되는 것을 확인할 수 있다.

Flow is reactive

Flow 는 Reactive 한 속성을 갖는다. 다른 Reactive Programming 라이브러리와도 호환이 될 수 있다.

Reactive 라이브러리들은 공통의 Specification 을 구현하도록 되어있는데 그 중 하나가 Publisher 라는 인터페이스이다. 코틀린에서는 Flow 가 다른 Reative 라이브러리들과 호환되도록 Flow 를 Publisher 로 변환하거나 Publisher를 Flow 로 변환하는 등의 함수를 가지고 있다.

Why Flow?

그렇다면 다른 RxJava 와 같은 Reactive 라이브러리를 사용하지 않고 Kotlin Flow 를 만들어야 하는 이유는 무엇이었을까?

기존 Reactive 라이브러리에서는 데이터스트림에 mapping 이나 filtering 을 수행할 경우 Asynchronous 한 작업을 수행하려면 추가적인 operator 를 사용하거나 간단한 코드로는 아예 불가능한 경우도 있었다.

그러나 코틀린 Flow 의 데이터스트림에서는 mapping 이나 filtering 을 수행하는 와중에도 suspend 라는 특성을 이용해 Asynchronous 한 작업을 수행해줄 수 있는 장점이 있다.

Operator avoidance

이러한 특성 때문에 코틀린 Flow 에서는 다른 Reactive 한 라이브러리와는 다르게 많은 수의 Operator 를 만들 필요가 없었다.

Flow Under The Hood

Flow 내부 구현을 살펴보면 다음과 같이 간단한 2개의 인터페이스로 구성되어 있는 것을 확인할 수 있다.

Flow 의 실행 흐름은 다음과 같다. Flow 에 collect 함수를 호출하면 flow 빌더에 전달한 람다가 실행된다. 이 람다에서 값을 emit 하는 경우 collect 함수에 전달한 람다가 실행되고, 이 람다 함수의 실행이 종료되면 다시 flow 빌더 람다의 다음 라인이 실행된다. flow 빌더에 전달한 람다의 모든 코드가 실행된 경우 collect 함수가 종료된다.

눈여겨 볼것은 collect 함수에 전달한 람다나 flow 빌더에 전달한 람다 내부에서 delay 와 같은 suspend 함수를 호출할 수 있다는 점이다. 이때 또 중요한 것은 collect 함수 람다 내부에서 delay 를 실행하면 flow 빌더의 다음값 emit 실행 시점도 같이 밀려난다는 점이다. flow 빌더에 전달한 람다는 collect 를 실행한 동일한 코루틴에서 실행된다.

Kotlin Flow Plays Scrabble

simple design 은 성능 향상을 이끌었다.

Flow is Asynchronous Yet Sequential

Flow 는 앞서 소개한것처럼 Asynchronous 한 동작을 수행할 수 있다. 그런데 Asynchronous 한 동작들을 Sequential 하게 수행한다. Asynchronous 한 작업을 마칠때까지 다음 작업을 실행할 수 없다.

이러한 동작이 효율적이지 않을 경우도 있다. 만약 collect 를 호출한 부분에서 값을 다 처리하기 전에 다음 값을 미리 계산해두고 emit 할 준비를 하고 싶은 경우도 있을것이다. 이런 경우를 해결하기 위해 Single Coroutine 을 사용하는 Flow 에서 Multiple Coroutine 을 사용하는 Flow 로의 확장이 필요하다.

Going Concurrent With A Flow

Flow 의 collect 와 emit 을 Concurrent 하게 수행하기 위해서 복잡한 코드를 작성해야만 하는 것은 아니다. 간단하게 flow 빌더와 collect 함수 호출 사이에 buffer 라는 연산자를 추가해주면 된다. 이렇게 하여 emit 을 호출하는 부분과 collect 하는 부분이 별도의 코루틴을 통해 concurrent 하게 실행될 수 있다.

내부적으로는 channel 을 이용해서 emit 하는 부분과 collect 하는 부분에 데이터 통신이 이루어지는데 Flow 사용자는 channel 을 직접 다뤄줄 필요가 없고 Flow api 내부적으로 모두 처리해주기 때문에 안전하게 작업을 처리해줄 수 있다.

Flow Execution Context

UI 애플리케이션 같은 경우에 UI 를 업데이트하는 코드가 어느 Thread 에서 실행될 것인가를 판단하는 것이 중요할 수 있다. 따라서 이번에는 Flow 가 실행되는 Context 에 대해서 알아본다.

기본적으로 flow 내부에 계산은 collect 를 호출한 코루틴의 context 에서 실행된다. collect 와 emit 이 단순 함수 호출이라는 것을 생각해볼때 이는 당연하다.

그런데 만약 Flow 내부에서 실행할 작업이 CPU intensive 한 작업이라서 collect 를 호출한 context 와는 다른 context 에서 실행되기를 원하는 경우에는 어떻게 해야할까? 이럴때는 flowOn 이라는 함수를 통해서 flow 가 값을 계산할때 사용할 context 를 명시해줄수 있다.

중요한 것은 flowOn 을 사용해서 upstream 이 실행되는 context 를 변경했다 하더라도 collect 에 전달한 람다의 실행 context 는 변하지 않는다는 점이다. upstream flow 가 down stream 에 영향을 주지않는다는 Context Preservation 개념을 알아둘 필요가 있다.

Flow In Reactive UI

UI 애플리케이션의 대표적인 패턴을 다음과 같이 구현할 수 있다.

인텐트를 줄이기 위해 다음과 같은 확장함수를 사용할 수 있다.

Managing Lifetime

Rx 를 사용할때는 작업 취소에 대한 처리를 직접 해줘야 했다. 이런 처리를 까먹을 가능성도 있었다.

그러나 Flow 에서는 반드시 CoroutineScope 를 명시하게 되어있기 때문에 이런 작업 취소를 까먹을 염려도 없고, Scope 취소가 자동으로 되어지는 UI Framework 가 있는 경우 단순히 Flow 가 실행될 CoroutineScope 를 명시해주기만 하면 작업 취소를 간단하게 구현할 수 있다.