[안드로이드] Coroutine Flow

https://developer.android.com/kotlin/flow?hl=ko

Android에서의 Kotlin 흐름  |  Android Developers

Flow는 여러 값을 순차적으로 내보낼 수 있는 기능이다.

Flow은 코루틴 기반으로 빌드되며, 비동기식으로 계산할 수 있는 데이터 스트림의 개념이다.

 

데이터 스트림에는 3가지 항목이 있다.

• 생산자 : 스트림에 추가되는 데이터를 생산
• 중개자 : 데이터 수정 ( 선택사항 )
• 소비자 : 데이터 사용

출처: 안드로이드 공식문서

 

1. 서버에서 데이터를 생산자가 받고

2. 중개자가 데이터를 가공한 뒤 (선택사항)

3. 소비자에게 넘겨주는 비동기식 기능이라고 보면 된다.

 

Flow가 필요한 이유

간단한 데이터 처리라면 Flow를 쓰지 않아도 메모리와 쓰레드 사용에 문제가 없을 것이다.그러나 대용량의 데이터 처리가 필요한 경우 비동기는 필수적이며, 메인 쓰레드에서 작업하기에는 리스크가 매우 크다.따라서 코루틴과 함께 사용할 수 있어 메인 쓰레드가 아닌 곳에서 데이터 처리가 비동기로 가능하기 때문에 유용하다고 할 수 있다.

또한, Flow에는 Backpressure를 처리할 수 있는 기능이 탑재되어 있어 효과적으로 관리할 수 있다.

 

Backpressure 란?

만일 내보내는 데이터를 소비자가 감당하지 못한다면 어떻게 될까?

예를 들어 초당 100개의 데이터가 내보내진다고 가정하고, 소비자는 데이터를 1초에 한개 사용한다고 할 때, 소비자는 넘어오는 데이터를 감당하지 못하게 된다.

그러면 데이터는 계속 쌓이게 되고 최악의 경우 메모리 문제 또는 애플리케이션 충돌이 발생할 수 있다.

이를 Backpressure 라고 한다.

 

사용 방법

빌드 (생산자)

기본

// Room
@Quert("~")
fun getData(): Flow<T>

// Retrofit
@FormUrlEncoded
@POST("/accounts/signup/")
fun getData(@Field("id") username:String): Flow<T>

다음과 같이 사용하면 자동으로 데이터가 Flow로 넘어간다.

 

Flow는 suspend fun과 같은 정지함수이기 때문에 suspend를 붙여주지 않는다.

 

수동으로 또는 다른 방법으로 빌드를 하고 싶은 경우

fun getData() = flow{
    val it = Api.getData()
    emit(it)
}

// 공식문서 예제
val latestNews: Flow<List<ArticleHeadline>> = flow {
    while(true) {
        val latestNews = newsApi.fetchLatestNews()
        emit(latestNews) // Emits the result of the request to the flow
        delay(refreshIntervalMs) // Suspends the coroutine for some time
    }
}

emit(): Flow에 데이터를 집어넣는 메서드

 

빌드에 사용되는 다양한 메서드

https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/

Flow

flowOf(...): 고정된 값 세트에서 흐름을 생성

asFlow(): 확장 기능은 다양한 유형을 흐름으로 변환

flow { ... }: 방출 함수에 대한 순차적 호출에서 임의의 흐름을 구성

ChannelFlow { ... }: 송신함수 에 대한 잠재적인 동시 호출에서 임의의 흐름을 구성

MutableStateFlow 및 MutableSharedFlow: 해당 생성자 함수를 정의하여 직접 업데이트할 수 있는 hot flow를 생성

 

변환 (중개자)

데이터를 변환할 때는 map{ ... } 과 filter{ ... }를 통해 변환이 가능하다.

// map
val changeData: Flow<T> = Repository.getData().map{ it -> it.toString() }

//filter
val changeData: Flow<T> = Repository.getData().filter{ it -> it.id == 1 }

 

예외처리

 

예외처리는 catch를 통해 가능하다.

// map
val changeData: Flow<T> = Repository.getData()
                                    .map{ it -> it.toString() }
                                    .catch{ e ->
                                    	Log.e("Error", e.toString())
                                        //TODO something
                                    }

//filter
val changeData: Flow<T> = Repository.getData()
                                    .filter{ it.id == 1 }
                                    .catch{ e ->
                                    	Log.e("Error", e.toString())
                                        //TODO something
                                    }

catch 는 업스트림에 대한 오류를 처리한다.

 

업스트림: (catch{...} 기준 업스트림) catch{...} 보다 위에 있는 Repository.getData().filter{...}, Repository.getData().map{...}

다운스트림: catch{...}보다 아래 있는 것

 

flowOn( context: CoroutineContext )

- flowOn 업스트림의 실행 코루틴 설정

 Repository.getData()
           .map{ it -> it.toString() } // run IO Thread
           .flowOn(Dispatchers.IO) 
           .filter { ... } // run Main Thread
           .flowOn(Dispatchers.Main)

 

map 연산은 IO 스레드에서, filter 연산은 Main 스레드에서 실행된다.

 

 

수집 (소비자)

수집은 보통 UI에서 사용된다.

Flow는 suspend fun과 같은 정지 함수로 코루틴 내부에서 수집되어야 한다.

 

수집은 collect 메서드를 사용

lifecycleScope.launch{
	changeData.collect{
		adapter.setData(it)
	}
}

 

이처럼 collect로 호출할때만 데이터를 수집한다.

 

이를 Cold Flow, Cold Stream이라고 한다.

Flow는 기본적으로 LifeCycle을 인지하지 못하기 때문에 UI가 안나오는 상황에서는 수집을 종료해줘야한다.

그러나 LifeCycle을 인지하지 못하기 때문에 이에 대해 다음과 같이 작업해줘야 한다.

 

1. LiveData

 

우리는 LiveData를 사용하여 데이터를 관찰, 변경이 일어나면 UI에 적용시켜주었다.

Flow에는 LiveData로 변환시켜주는 기능이 있기 때문에 Flow로 수집, 변환 후 LiveData로 변환하여 기존의 LiveData처럼 사용할 수 있다.

 

val liveData = Repository.getData().asLiveData()

 

Repository.getData() 는 Flow<T>를 반환하지만 asLiveData()를 사용하여 LiveData로 변환이 가능하다.

 

2. repeatOnLifecycle, flowWithLifecycle

해당 함수는 suspend 함수로 Lifecycle.State를 파라미터로 받는다.

또한 Lifecycle을 인지하여 동작하게 되는데 구조는 다음과 같다.

 

repeatOnLifecycle

lifecycleScope.launch{
	repeatOnLifecycle(Lifecycle.State.STARTED){
		changeData.collect{
			adapter.setData(it)
		}
    }
}

 

flowWithLifecycle : 수집하는 데이터가 하나 뿐일 경우 사용

 lifecycleScope.launch{
    changeData.flowWithLifecycle(lifecycle, State.STARTED)
              .collect{ adapter.setData(it) }
}

 

 

 

 

StateFlow

StateFlow는 현재 상태와 새로운 상태 업데이트를 내보내는 관찰 가능한 state-holder Flow 이다.

Flow 버전 LiveData라고 봐도 무방하다.

 

사용 방법은 다음과 같다.

 

1. stateIn()

val result: StateFlow<T> = Repository.getData()
                                     .stateIn(
                                     	initalValue = Result.Loading,
                                        scope = viewModelScope,
                                        started = WhileSubscribed(5000)
                                     )

 

stateIn()에는 3가지 파라미터가 필요하다.

stateIn( scope: CoroutineScope, started: SharingStarted, initalValue: T )

• scope: stateFlow가 동작할 코루틴 스코프
• started: 언제부터 구독을 할지
• initalValue: 초기 값

stateFlow에서는 started에 WhileSubscribed(5000)을 권장하고 있다.

WhileSubscribed()는 stateFlow의 수집이 종료되었을 시점으로부터 몇 초 뒤에 중단하는지 정해주는 기능으로 5000은 5초를 뜻한다.

 

수집이 종료되는 시점은 보통 홈으로 넘어가 뷰가 중단되는 경우인데 이 시점으로 5초가 지나면 뷰가 중단되었다는 확신으로 Flow를 중단시키는 

 

2. MutableStateFlow

val _stateFlow = MutableStateFlow<T>()
val stateFlow: StateFlow<T> = _stateFlow

init{
    viewModelScope.launch{
    	val getData = repository.getData()
    	_stateFlow.emit(getData)
    }
}

 

해당 방법은 LiveData와 마찬가지로 MutableStateFlow에 받아온 값을 넣어 이를 다시 StateFlow에 적용시켜 사용하는 방법이다.