[Kotlin][Coroutine] Flow의 이해

 

 

Coroutine의 Flow는 비동기적으로 계산해야할 데이터의 스트림이며, 코루틴 상에서 리액티브 프로그래밍을 가능하게 한다. 본격적으로 flow에 대해 알아보기 전에, 리액티브 프로그래밍이 무엇인지부터 알아보자.

 

리액티브 프로그래밍

• 데이터 발행자(Publisher)가 데이터를 발행하면, 그 구독자가 데이터를 처리하는 프로그래밍을 리액티브 프로그래밍이라고 한다.

fun main(){
	val publisher : PublishSubject<String> = PublishSubject.create()

	publisher.subsrcibe { it-> println(int)}

	publisher.onNext("count:1")
	publisher.onNext("count:2")
	publisher.onNext("count:3")
}

• PublishSubject: Hot Stream을 생성한다. 즉, 구독자가 없어도 데이터를 즉시 방출하기 시작한다
• subscribe: 이 메서드는 게시자와 구독자(이 경우 println 함수) 간의 연결을 설정한다
• onNext: 이 메서드는 데이터("count:1", "count:2", "count:3")를 구독자가 수신할 수 있도록 스트림에 푸시한다

 

기존 명령형 프로그래밍은 컴퓨터가 사용자에게 동작을 일일이 알려줘야했다면, (ex. mainBtn1.setOnClickListener()~) 위 예제와 같이 발행자와 이벤트 발생시의 동작을 정의하면 해당 발행자에서 데이터가 발행되는대로 이를 구독해 동작을 수행할 수 있다. 

 

 

Flow

앞서 언급했듯이 Flow는 비동기적으로 계산해야할 데이터의 스트림이며, 코루틴 상에서 리액티브 프로그래밍을 가능하게 한다. Flow 인터페이스 자체는 떠다니는 원소들을 모으는 역할을 하고, 끝에 도달할 때까지 값을 처리한다. 그렇다면 flow는 어떨 때 유용할까 ?

 

여러개의 값을 반환하는 함수가 있고 사용자는 원소를 계산할때마다 바로바로 하나씩 반환하기를 원하는 상황이라고 해보자. 그러면 코틀린 사용자라면 시퀀스를 떠올릴 것이다. 시퀀스에 대해서는 따로 자세히 다루고, 간단하게 정리하면 iterable과 유사하지만 LAZY한 처리를 한다는 것이 특징이다. 

 

이때 시퀀스의 최종 연산은 중단 함수가 아니기 때문에 시퀀스 빌더 내부에 중단점이 있다면 값을 기다리는 스레드가 블로킹된다. 이때 스레드의 블로킹은 따로 설명하지 않아도 위험하며, 예기치않은 상황을 유발할 수도, 성능 저하의 원인이 될 수 있다. 그렇다면 시퀀스말고 다른 걸 도입해 이 문제를 해결해야 한다. 그렇다면 조건을 정리해보면 다음과 같다.

 

• 연산이 중단 함수여야한다
• 예외처리가 가능해야한다
• 값(데이터)를 가질 수 있어야한다

시퀀스가 첫번째 조건을 만족하지 못했다면, flow의 연산자는 중단 함수이며 스레드를 블로킹하는 대신 코루틴을 중단시키므로 합격이다. 이게 리액티브 프로그래밍 flow가 사용되는 이유이다. 

 

 

 

Flow의 특징

• flow의 최종연산은 스레드를 블로킹하는 대신 코루틴을 중단시킨다
• flow는 CoroutineContext를 활용하고 예외를 처리하는 코루틴 기능 또한 제공한다
• flow는 취소 가능하다
• flow는 구조화된 동시성을 갖췄다
• flow빌더는 중단 함수가 아니며 어떠한 스코프도 필요로 하지 않는다(CoroutineScope 불필요)
• flow의 최종 연산은 중단 가능하며, 연산이 실행될때 부모 코루틴과의 관계가 생긴다

fun usersFlow(): Flow<String> = flow {
    repeat(3) {
        delay(1000L)
        val ctx = currentCoroutineContext()
        val name = ctx[CoroutineName]?.name
        emit(("User$it : $name"))
    }
}

suspend fun main(){
    val users = usersFlow()
    withContext(CoroutineName("HELLO FLOW")) {
        val job = launch { users.collect { println(it) } }
        launch {
            delay(2500L)
            println("2500L 지남 - Cancel Job : flow도 취소 -> User2 안나와")
            job.cancel()
        }
    }
}

 

 

위 예제를 실행시켜보면 다음과 같은 출력이 나온다

User0 : HELLO FLOW
User1 : HELLO FLOW
2500L 지남 - Cancel Job : flow도 취소 -> User2 안나와

 

3번의 반복을 통해서 유저는 0~2까지 존재하지만 2500L 이후 코루틴을 취소하면, flow 또한 취소 되므로 유저 0~1까지만 출력된다.

 

또, 중단 함수는 코루틴 스코프 내에서만 호출될 수 있는데 flow { } 를 보면 CoroutineScope 내에서 호출되지 않기 때문에 중단함수가 아님을 확인할 수 있다. 다만, flow { } 블록 내부가 중단 함수일 수는 있고, collect{ } 와 같은 연산자가 중단 함수이다. 

 

 

flow 사용 예

데이터 스트림을 사용하면, 대부분은 데이터가 필요할 때마다 요청한다. 

• 웹소켓이나 RSocket 알림과 같이 서버가 보낸 이벤트를 통해 전달된 메시지를 받는 경우
• 텍스트 입력 또는 클릭과 같은 사용자 액션이 감지된 경우
• 센서 또는 위치나 지도와 같은 기기 정보 변경을 받는 경우
• 데이터베이스의 변경을 감지하는 경우

이렇게 데이터나 이벤트, UI의 변화를 감지하는 경우에 활용되기 좋다. 

 

 

 

이렇게 오늘은 flow의 등장 배경과 장점, 사용 예시에 대해 알아보았다면 다음엔 flow를 직접 구현하는 방법과 빌더, 연산자와 같은 개념의 활용에 대해 더 자세히 알아보자. 

 

 

 

 

참고자료

 

마르친 모스카와 - Kotlin Coroutines : Deep Dive 19장

Android에서의 Kotlin 흐름  |  Android Developers