Android/Kotlin Coroutine StateFlow, SharedFlow, Channel 예제 및 특징

출처 핀터레스트

 

3번째 flow 예제 공부를 마쳤다.

 

stateflow, sharedflow, channel에 대한 예제를 꾸렸지만,

 

모두 클릭 리스너를 달아 데이터를 가져오는데 그친 간단한 예제이지만,

 

기능보다는 각각의 특징과 공부한 내용들로 주를 이루었다.

 

우선 코드를 보자.

 

MainViewModel.kt

 

class MainViewModel : ViewModel() {

    //StateFlow
    private val _uiState = MutableStateFlow<UiState>(UiState.Loading)
    val uiState: StateFlow<UiState> = _uiState.asStateFlow()

    //SharedFlow
    //설정할 수 있다는 것만 보여주기 위함
    //아래 설정은 다 기본값
    private val _uiShared = MutableSharedFlow<UiState>(
        replay = 0,
        extraBufferCapacity = 0,
        onBufferOverflow = BufferOverflow.SUSPEND
    )
    val uiShared: SharedFlow<UiState> = _uiShared.asSharedFlow()


    //Channel
    //Channel.BUFFERED 버퍼 설정이 가능하다
    //Channel.BUFFERED는 시스템이 정한 버퍼값 64개를 가지고 있는 상태로 설정한다는 뜻이다.
    val uiChannel = Channel<UiState>(Channel.BUFFERED)

    init { getData() }

    fun getData() {
        viewModelScope.launch {
            createBlogList().collect {
                UiState.Success(it).let { value ->
                    _uiState.value = value
                    //SharedFlow는 value가 없음
                    _uiShared.emit(value)
                    uiChannel.send(value)
                }
            }
        }
    }
}

 

StateFlow와 SharedFlow부터 살펴보겠다. 우선 state와 shared는 livedata와 mutablelivedata와 비슷한 형태를 띤다.

 

StateFlow는 처음에 초깃값을 가지고, SharedFlow는 초깃값을 가지지 않아도 된다.

 

게다가 원한다면, 다음과 같은 속성을 정할 수 있다.

* StateFlow처럼 최신값을 보내는 것과는 다르게 내보낼 이전의 값 수를 구성할 수 있다. (replay)
* emit된 데이터를 저장(cache)할 버퍼의 갯수를 지정할 수 있습니다. (extraBufferCapacity)
* buffer가 다 찼을때 동작을 정의할 수 있습니다. (onBufferOverflow)

그리고 아래에는 채널이 있다.

 

채널은 다음과 같이 객체를 만들고, 선택사항으로 버퍼의 사이즈를 지정할 수 있다.

 

//Channel
//Channel.BUFFERED 버퍼 설정이 가능하다
//Channel.BUFFERED는 시스템이 정한 버퍼값 64개를 가지고 있는 상태로 설정한다는 뜻이다.
val uiChannel = Channel<UiState>(Channel.BUFFERED)

Channel.BUFFERED는 시스템이 정한 버퍼 값 64개를 가지고 있는 상태로 설정한다는 뜻이다.

 

---

 

처음 시작할 때 getData() 메서드를 호출하여 블로그 리스트를 가져온다.

 

가져온 리스트들을 각각의 StateFlow, SharedFlow, Channel에 넣어주었다.

 

fun getData() {
    viewModelScope.launch {
        createBlogList().collect {
            UiState.Success(it).let { value ->
                _uiState.value = value
                //SharedFlow는 value가 없음
                _uiShared.emit(value)
                uiChannel.send(value)
            }
        }
    }
}

 

추가적으로 StateFlow는 value emit()으로 방출하는 방법이 둘 다 존재하지만,

 

SharedFlow는 emit() 메서드만 있고, value 속성이 없다.

 

channel은 send로 값을 송신한다.

 

ui단으로 넘어가 보자.

 

MainActivity.kt

 

class MainActivity : AppCompatActivity() {

    private val vm by viewModels<MainViewModel>()

    private lateinit var binding: ActivityMainBinding

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        binding = ActivityMainBinding.inflate(layoutInflater)
        setContentView(binding.root)
        lifecycleScope.launch {
            repeatOnLifecycle(Lifecycle.State.STARTED) {
                //StateFlow
                launch {
                    vm.uiState.collect { state ->
                        if (state is UiState.Success<*>) {
                            Log.d("uiState : ", state.data.toString())
                        }
                    }
                }
                //SharedFlow
                launch {
                    vm.uiShared.collect { state ->
                        if (state is UiState.Success<*>) {
                            Log.d("uiShared : ", state.data.toString())
                        }
                    }
                }

              

                launch {
                    vm.uiChannel
//                        .consumeAsFlow() //find, map, filter, first 같은 함수들을 사용하려면 cosumeAsFlow()
                    //함수를 이용해 flow로 전환하고 사용해야한다.

//                        .receive() //리시브로 받는 방법
                    //이 방법을 이용하려면 채널의 데이터를 보내지 않거나, 받지 않으면 close()메서드를
                    //통해 채널을 종료해주어야한다. close() 메서드 이후 send()나 receive() 메서드를 호출하면,
                    // ClosedReceiveChannelException 예외가 발생됩니다.

//                        .consumeEach {  } //consumeEach와 아래 코드처럼 for문을 돌리는 방법은
                    //실행하는데 매우 유사한 방법이지만,
                    //consumeEach는 실행중 에러가 발생하면, 채널 자체가 close되지만,
                    //for문을 돌리는 방법은 for문만 중단이 되고, channel은 종료되지 않는다.
                    //이 채널을 사용하고 있는 다른 곳은 정상적인 동작을 이어 간다는 것이다.
                    //가이드 문서에스는 for-loop를 완벽하게 안전한 것이라 표현함.
                    //데이터를 recieve할때에는 for-loop사용 권장

                    for(x in vm.uiChannel) {
                        if (x is UiState.Success<*>) {
                            Log.d("uiChannel : ", x.data.toString())
                        }
                    }
                }
            }
        }
        clickListener()
    }

    private fun clickListener() {
        //데이터 가져오기
        binding.sameData.setOnClickListener { vm.getData() }
    }
}

차근차근 위에서부터 보겠다.

 

lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) { 
    ...
        }
    }

 

suspending 함수들을 다루기 위해 lifecycleScope로 감싸고,

 

내부에는 repeatOnLifeCycle(Lifecycle.State.STARTED) 블록으로 한번 더 감싸주었다.

 

감싸준 이유는 아래와 같다.

 

    flow는 livedata처럼 lifecycle의 흐름대로 자동으로 종료되거나 재시작되지 않습니다.

만약 앱이 백그라운드로 가더라도 계속 emit 된 값을 collect 합니다.

그래서 repeatOnLifecycle Lifecycle.State.STARTED를 사용합니다.
STARTED를 보게 되면,
activity가 onstart 되는 시점이나 onPause 되는 시점에 새로운 coroutine으로 시작되며, onStop시점에 cancel 됩니다.

 

이제 각각의 StateFlow, SharedFlow, Channel들을 감싸줄 launch 블록을 만들어주고,

 

각각의 특징들을 보겠다.

 

StateFlow & ShareFlow

//StateFlow
launch {
    vm.uiState.collect { state ->
        if (state is UiState.Success<*>) {
            Log.d("uiState : ", state.data.toString())
        }
    }
}
//SharedFlow
launch {
    vm.uiShared.collect { state ->
        if (state is UiState.Success<*>) {
            Log.d("uiShared : ", state.data.toString())
        }
    }
}

둘 다 들어온 값에 대한 collect를 하여 값을 수집한다.

 

다를 게 없어 보이지만, 이 둘은 명확한 차이점이 있다.

 

우선 공통점부터 보자.

 * StateFlow와 ShareFlow의 공통점은 HotStream이다.
 * HotStream이란 한 flow에서 emit을 할때 여러 구독자(collect)가 동시에 데이터를 구독할 수 있는 스트림을 말한다.
 * 예시로는 라디오 방송국이 있다. 우리가 라디오 방송국에서 나오는 라디오 주파수를 맞추게 되면, 동시에 여러 사람들이
 * 라디오를 들을 수 있는 상황과 같다고 보면 된다.
 * HotStream은 중간에 collect를 하면 처음부터 값을 받을 수도 있지만, 중간에 받게 되면 중간부터 데이터를 받게 된다.
 * (라디오처럼)
 * 추가적을 ColdStream은 무엇일까?
 * ColdStream은 하나의 구독자만을 위한 스트림이다. HotStream이 외부에서 데이터를 받고 왔다면, ColdStream은
 * 내부에서 데이터를 생성하여 하나의 구독자에게 전달한다.
 * 이 말은 여러 구독자가 있어도 각각의 인스턴스를 가진 스트림을 부여하기 때문에 독립성을 띄고 있다.
 * 하나의 인스턴스로 공유하는 HotStream과 다른 점이다.

여기까지는 두 HotStream에 대한 공통점이다.

 

이제 차이점에 대한 정리도 보자.

* 바로 conflate이다.
* conflate란 사전적으로 두개의 값을 하나로 병합한다는 의미이다.
* 하지만 flow에서는 약간 다르게 쓰인다.
* flow에서는 collect로 넘어온 값이 이전값과 동일하다면 스킵(병합)시키는 기능으로 통한다.
* StateFlow에는 내부적으로 conflate가 구현되어 있다.
* 고로, 최신값이 이전값과 동일하다면, 값이 넘어오질 않는다.
* 하지만, SharedFlow는 conflate가 구현되어 있지않다.
* conflate() 메서드를 통해 conflate 기능을 따로 구현해주어야 한다.

과연 정말 그럴까?

 

앱이 시작하면 데이터를 한번 불러오게 되는데,

 

이 상황에서 같은 값을 한번 더 불러오는 viewModel에 getData() 메서드를 한번 더 호출해보자.

 

처음 앱을 실행할 때는 이런 식으로 값을 잘 가져온다.

 

아래는 getData()를 호출할 때이다.

 

 

로그에는 ShareFlow에만 데이터가 온 것으로 찍혔다.

 

그렇다. 위에 설명처럼 StateFlow는 자체적으로 conflate가 구현되어 있기 때문에,

 

이전 값과 최근 값이 같다면, collect를 통해 데이터가 넘어오지 않는다. 

 

다음은 두 flow의 각각의 특징이다.

* StateFlow 특징
* 초깃값이 필요하다. 생성자에 초기값을 반드시 명시해야한다. (항상 값을 가지고 있고, 오직 한 가지 값을 가진다.)
* emit()과 value 속성 이용가능
* collector의 수에 관계없이 항상 구독하고 있는 것의 최신 값을 받는다.
* hot stream이다.
* collect를 활성화한 시점부터 방출된 데이터를 가져온다.
* 업데이트 된 후에 값을 반환하고 동일한 값을 반환하지 않습니다.(conflate)

* SharedFlow 특징
* 초깃값이 필요없습니다.
* value 사용 불가, emit()만 사용가능
* StateFlow의 기본적인 원형의 가까운 형태로 볼 수 있다.
* StateFlow처럼 최신값을 보내는 것과는 다르게 내보낼 이전의 값 수를 구성할 수 있다. (replay)
* emit된 데이터를 저장(cache)할 버퍼의 갯수를 지정할 수 있습니다. (extraBufferCapacity)
* buffer가 다 찼을때 동작을 정의할 수 있습니다. (onBufferOverflow)
* hot stream입니다.
* 업데이트하면 이전 값이 동일하든 말든 값이 반환됩니다. (conflate 따로 설정해야함.)

 

다음으로 볼 것은 Channel이다,

 

Channel

 

launch {
                    vm.uiChannel
//                        .consumeAsFlow() //find, map, filter, first 같은 함수들을 사용하려면 cosumeAsFlow()
                    //함수를 이용해 flow로 전환하고 사용해야한다.

//                        .receive() //리시브로 받는 방법
                    //이 방법을 이용하려면 채널의 데이터를 보내지 않거나, 받지 않으면 close()메서드를
                    //통해 채널을 종료해주어야한다. close() 메서드 이후 send()나 receive() 메서드를 호출하면,
                    // ClosedReceiveChannelException 예외가 발생됩니다.

//                        .consumeEach {  } //consumeEach와 아래 코드처럼 for문을 돌리는 방법은
                    //실행하는데 매우 유사한 방법이지만,
                    //consumeEach는 실행중 에러가 발생하면, 채널 자체가 close되지만,
                    //for문을 돌리는 방법은 for문만 중단이 되고, channel은 종료되지 않는다.
                    //이 채널을 사용하고 있는 다른 곳은 정상적인 동작을 이어 간다는 것이다.
                    //가이드 문서에는 for-loop를 완벽하게 안전한 것이라 표현함.
                    //데이터를 receive할때에는 for-loop사용 권장

                    for(x in vm.uiChannel) {
                        if (x is UiState.Success<*>) {
                            Log.d("uiChannel : ", x.data.toString())
                        }
                    }
                }

복잡해 보이지만, 하나씩 보겠다.

 

Channel을 소개할 때는 여러 곳으로 데이터를 던지고 받는데,

 

이때 사용하는 용어가 Fan-In(receive), Fan-out(send)이라고 한다.

 

위에 코드에서는 ViewModel에서 Fan-out(send)를 했으므로,

 

우리가 볼 부분은 Fan-In(receive)이다.

 

vm.uiChannel
//                        .consumeAsFlow() //find, map, filter, first 같은 함수들을 사용하려면 cosumeAsFlow()
                    //함수를 이용해 flow로 전환하고 사용해야한다.

//                        .receive() //리시브로 받는 방법
                    //이 방법을 이용하려면 채널의 데이터를 보내지 않거나, 받지 않으면 close()메서드를
                    //통해 채널을 종료해주어야한다. close() 메서드 이후 send()나 receive() 메서드를 호출하면,
                    // ClosedReceiveChannelException 예외가 발생됩니다.

//                        .consumeEach {  } //consumeEach와 아래 코드처럼 for문을 돌리는 방법은
                    //실행하는데 매우 유사한 방법이지만,
                    //consumeEach는 실행중 에러가 발생하면, 채널 자체가 close되지만,
                    //for문을 돌리는 방법은 for문만 중단이 되고, channel은 종료되지 않는다.
                    //이 채널을 사용하고 있는 다른 곳은 정상적인 동작을 이어 간다는 것이다.
                    //가이드 문서에는 for-loop를 완벽하게 안전한 것이라 표현함.
                    //데이터를 receive할때에는 for-loop사용 권장

 

여러 종류가 있는데 하나씩 보겠습니다.

 

channel 자체에서는 find, map, filter, first 같은 함수를 사용할 수 없어, (원래 있었지만, 현재 Deprecated 된 상태입니다.)

 

사용하려면 flow로 전환해야 사용이 가능하다.

 

이때, 사용하는 메서드가 consumeAsFlow() 메서드입니다.

 

 

간단히 send()로 데이터를 보내고, receive()로 받는 방법도 있습니다.

 

하지만 이 방법을 사용할 때는 채널의 데이터를 보내지 않거나, 사용하지 않으면, close() 메서드를 통해 닫아주어야 합니다.

 

close()를 사용한 이후에 send()나 receive()를 호출하면,

 

ClosedReceiveChannelException 예외가 발생됩니다.

 

그리고,

 

consumeEach()와 for-loop를 이용한 방법도 존재합니다.

 

두 방법은 기능상 차이점은 없지만,

 

consumeEach() 메서드를 사용 중에 에러가 발생하면,

 

채널 자체가 close 되고, 같은 채널을 사용하고 있는 다른 곳에도

 

채널이 종료되어 사용할 수 없게 되는 단점이 존재합니다.

 

하지만,

 

for-loop로 돌리게 되면, for문만 중단이 되고,

 

채널은 종료되지 않습니다.

 

채널을 사용하고 있는 다른 곳에서는 정상적으로 동작을 이어가는 것이지요.

 

 가이드 문서에는 for-loop를 완벽하게 안전한 것이라 표현하였고,

 

데이터를 receive 할 때에는 for-loop를 사용하라고 권장하였습니다.

 

자 그러면, channel의 특징을 정리한 글도 보겠습니다.

 

* Channel 특징
* flow가 단일방향으로 데이터를 던지고 받는 형식이라면,
* channel은 여러 방향에서 데이터를 던지고 받는 형식, 코루틴 끼리의 데이터를 전달해준다.
* 현재 위 코드 사용은 클릭리스너를 위한 방법이므로, 일반적으로는 코루틴 간의 데이터 통신을 위해 사용한다.
* 데이터 배출은 hot stream이어서 사용을 기다리지 않고, 한번에 배출한다.
* 쉽게 말해 한번 밖에 배출하지 않아서, flow처럼 여러곳에서 동일한 데이터를 받는 것은 불가능하다.
* 버퍼 사이즈나 형태를 지정할 수 있다.
* FIFO (first-in first-out) 선입선출식으로 동작하며, 하나의 채널이 독점하지 않고,
* 순차적으로 데이터를 가져간다.

 

channel 역시 conflate가 내부에 없어, send 된 값이 전 값과 비교하지 않고, 그대로 receive 됩니다.

 

지금까지 StateFlow, SharedFlow, Channel의 특징에 대한 정리였습니다.

 

아래 두 글 역시 이번 flow의 대한 정리를 한 것입니다.

 

2022.09.04 - [App Development/후맛집 프로젝트] - Android/Kotlin Flow를 이용하여 데이터 상태에 따라 버튼 활성화 비활성화 변경하기 TIL # 86

Android/Kotlin Flow를 이용하여 데이터 상태에 따라 버튼 활성화 비활성화 변경하기 TIL # 86

2022.09.06 - [App Development/후맛집 프로젝트] - Android/Kotlin Flow onCompletion, catch, collect 예제

Android/Kotlin Flow onCompletion, catch, collect 예제

 

전체 코드는 아래 깃헙에 올려두었습니다.

 

https://github.com/qjsqjsaos/FlowUses

GitHub - qjsqjsaos/FlowUses: 후맛집 프로젝트 -> Flow 활용 과제