안드로이드 Flow 학습 및 예제 -flow, collect, buffer, collectLast, StateFlow, SharedFlow, map, filter-

 

https://www.udemy.com/course/android-architecture-componentsmvvm-with-dagger-retrofit/learn/lecture/31674208#overview

 

예전에 들었던 Udemy 안드로이드 강의를 최근에 들어가보니 Compose 와 Flow + 코루틴 강의가 추가되었다 ㅎㅎ

꾸준히 버전업데이트도 하고 새로운 강의를 만들어주셔서 안드로이드 Jetpack 및 아키텍처 강의로 베스트셀러인 이유가 있는거 같다!

추천추천~

 

강의는 매우 짧게 예제위주로 구성되어있어서 간단하게 기록용 포스팅이다. (거의 개인적인 사견이 많이 들어감... 간단한 맛보기 공부라 부족한점 이해 바랍니다)

 

평소 비동기 코드(통신)에 주로 사용하는 리엑티브 프로그래밍 언어로 RxJava를 사용해왔는데 요즘은 구글에서 만든 코루틴과 함께 Flow API가 대체되어가고 있어서 이번 기회에 강의로 Flow 를 맛만(?) 보아봤다.

 

개념은 같은 리엑티브 프로그래밍인만큼 RxJava와 비슷하고 코드가 간결하고 좀 더 가볍(쉽다)다는 장점이 있는 것 같다.

더 자세한 내용은 공식문서를 참고

https://developer.android.com/kotlin/flow

Android의 Kotlin 흐름  |  Android 개발자  |  Android Developers

 

포스팅에 앞서 코루틴과 RxJava가 비슷하면서 용어가 조금 달라서

소비자(Cousumer) == 구독자

collect == 구독

등 좀 혼용해서 용어를 사용하는 점 참고바랍니다. ㅎㅎ

---

 

코루틴도 RxJava처럼 데이터를 생성하는 생성자(Producer)와 생성자로부터 데이터를 전달받아 처리하는 소비자(Consumer)로 나뉜다.

// 생성자 (Producer)
val myFlow = flow<Int> {
	for (i in 1.100){
    	emit(i)
        delay(1000L)
    }
}

// 소비자 (Consumer)
CoroutineScope(Dispatchers.Main).launch{
	myFlow.collect{
    	Log.i("MyTAG", "Current Value ->  $it")
    }
}

위 코드는 flow 빌더함수를 통해 생성자가 1초 간격으로 1~10까지 Int형 데이터흐름을 생성하게 되고 소비자는 메인 스레드에서 앞서 정의한 생성자에서 방출(emit)하는 데이터흐름을 collect를 통해 소비할 수 있다.

참고로 flow 블록을 사용한 Flow는 누군가 collect를 하기 전까지는 데이터를 생성하지 않는 Cold Stream이다. 이부분은 이후 StateFlow를 볼때 한번더 언급할 것이다.

 

---

 

 

// 생성자 (Producer)
val myFlow = flow<Int> {
	for (i in 1.100){
    	Log.i("MyTAG", "Emit Value ->  $it")
    	emit(i)
        delay(1000L)
    }
}

// 소비자 (Consumer)
viewModelScope.launch{
	myFlow.collect{
    	Log.i("MyTAG", "Current Value ->  $it")
    }
}

// 결과
Emit Value -> 1
Current Value -> 1
Emit Value -> 2
Current Value -> 2
Emit Value -> 3
Current Value -> 3
Emit Value -> 4
Current Value -> 4
.
.
.

flow 블록을 통한 Flow는 collect로 데이터를 소비하기 시작하면 생성자에서 데이터를 생성 및 방출하게 되고 결과는 위와 같다.

 

 

 

 

// 생성자 (Producer)
val myFlow = flow<Int> {
	for (i in 1.100){
    	Log.i("MyTAG", "Emit Value ->  $it")
    	emit(i)
        delay(1000L)
    }
}

// 소비자 (Consumer)
viewModelScope.launch{
	myFlow.collect{
    	// delay 추가
    	delay(2000L)
    	Log.i("MyTAG", "Current Value ->  $it")
    }
}

// 결과
Emit Value -> 1
Current Value -> 1
Emit Value -> 2
Current Value -> 2
Emit Value -> 3
Current Value -> 3
Emit Value -> 4
Current Value -> 4
.
.
.

소비자쪽에 delay(2000L) 코드를 추가하여 소비속도가 생성속도보다 느려졌지만 결과는 같다.

왜냐하면 Flow는 이전에 방출한 데이터를 소비자가 소비하기전에 데이터를 방출하지 않기 때문이다.

 

 

이렇게 생성자가 소비자가 이전에 방출한 데이터를 처리할때까지 기다렸다 하나씩 전달하는 메커니즘은 생성자가 소비자에게 데이터를 다 전달해주지 못하고 끝나거나 과부하가 날수도 있다. 

이를 위해 RxJava와 비슷하게 방출대기중인 데이터들을 어떻게 다룰지 정하는 배압전략(backpressure)을 Flow에서도 제공한다. 

그 중 buffer 와 collectLast 기능에 대해 배웠다.

배압에 관련된 포스팅은 다음 글에서 재미있게 잘 설명하고 있는거 같다.

https://doublem.org/stream-backpressure-basic/

Stream Backpressure의 이해 | Doublem.org

 

 

buffer() 부터 보면 다음과 같다.

// 생성자 (Producer)
val myFlow = flow<Int> {
	for (i in 1.100){
    	Log.i("MyTAG", "Produced ->  $it")
    	emit(i)
        delay(1000L)
    }
}

// 소비자 (Consumer)
viewModelScope.launch{
	myFlow.buffer().collect{
    	// delay 추가
    	delay(2000L)
    	Log.i("MyTAG", "Consumed ->  $it")
    }
}

// 결과
Produced -> 1
Produced -> 2
Consumed -> 1
Produced -> 3
Produced -> 4
Consumed -> 2
Produced -> 5
Produced -> 6
Consumed -> 3
Produced -> 7
Produced -> 8
Consumed -> 4
Produced -> 9
Produced -> 10
Consumed -> 5
Consumed -> 6
Consumed -> 7
Consumed -> 8
Consumed -> 9
Consumed -> 10

생성자가 소비자의 데이터 처리에 신경쓰지 않고 블로킹없이 데이터를 생성하는 것을 볼 수 있다. 또한 소비자는 나중에라도 전달받은 데이터를 순차적으로 모두 소비하는 것도 볼 수 있다.

 

 

---

 

 

만약 가장 최근에 생성한 데이터만 소비하고자하면 어떻게 하면 좋을까? 그럴때는 collect 대신 collectLast를 사용하면 된다.

// 생성자 (Producer)
val myFlow = flow<Int> {
	for (i in 1.100){
    	Log.i("MyTAG", "Produced ->  $it")
    	emit(i)
        delay(1000L)
    }
}

// 소비자 (Consumer)
viewModelScope.launch{
	myFlow.collectLast{
    	// delay 추가
    	delay(2000L)
    	Log.i("MyTAG", "Consumed ->  $it")
    }
}

// 결과
Produced -> 1
Produced -> 2
Produced -> 3
Produced -> 4
Produced -> 5
Produced -> 6
Produced -> 7
Produced -> 8
Produced -> 9
Produced -> 10
Consumed -> 10

마지막 10 데이터만 소비하는 결과를 볼 수 있었다.

 

 

---

 

 

계속 같은말을 반복하지만 ㅎㅎ RxJava처럼 Flow 에서도 역시나 필터 함수기능들을 제공한다.

그 중 가장 대표적인 map과 filter를 사용해봤다.

// 생성자 (Producer)
val myFlow = flow<Int> {
	for (i in 1.100){
    	Log.i("MyTAG", "Produced ->  $it")
    	emit(i)
        delay(1000L)
    }
}

// 소비자 (Consumer)
viewModelScope.launch{
	myFlow.filter{  // 말 그대로 조건에 맞는 필터링한 데이터만 통과됨
    	count -> count % 3 = 3
    }
    .map { // 데이터를 가공함
    	it -> ("MAP " + it)
    }
    .collect{
    	Log.i("MyTAG", "Consumed ->  $it")
    }
}

// 결과
Produced -> 1
Produced -> 2
Produced -> 3
Consumed -> MAP 3
Produced -> 4
Produced -> 5
Produced -> 6
Consumed -> MAP 6
Produced -> 7
Produced -> 8
Produced -> 9
Consumed -> MAP 9
Produced -> 10

filter 로 데이터가 조건에 맞는거만 필터링되고 map으로 데이터를 가공할 수 있다는 것을 확인할 수 있었다.

 

 

 

---

 

 

RxJava와 마찬가지로 Flow에도 Cold Hot 개념이 있다.

지금까지한 flow 빌더함수(블록)을 활용한 Flow는 데이터를 소비자가 collect (RxJava로 치면 구독) 하기전까지 데이터를 생성하지 않는 Cold Stream이다. 또한 구독자(소비자)가 오로지 1명 뿐이라는 특징이 있다.

StateFlow 와 SharedFlow 는 소비자의 collect(구독)과 상관없이 늘 동작하는 Hot Stream이고 한 명 이상의 구독자(소비자)를 가질 수 있다.  이 둘에 대한 특징은 밑의 공식문서를 참고하자!

https://developer.android.com/kotlin/flow/stateflow-and-sharedflow

StateFlow 및 SharedFlow  |  Android 개발자  |  Android Developers

 

 

StateFlow 부터 살펴보면, 

기존에 사용하던 LiveData 와 비슷하고 비슷하게 사용하면 되었다.

하지만 비슷하지만 몇가지 차이점도 있다.

첫번째로,

LiveData는 초깃값(초기상태)을 가지고 있지 않으므로 null check가 필요하다.

StateFlow는 초깃값을 필요(필수)하므로 null check가 필요하지 않다. 

두번째로,

LiveData는 View가 Stopped 상태로 갈때 자동으로 구독자(소비자)를 해제해준다.

StateFlow는 기본으로 LiveData처럼 동작하지 않으므로 lifecycleScope 에서 이것을 해줘야한다 . 

세번째로,

LiveData는 안드로이드 프레임워크에 종속되서 안드로이드개발에서만 사용할 수 있다.

StateFlow는 코틀린 언어 종속으로 코틀린 멀티플랫폼 프로젝트에서 다양하게 사용이 가능하다 . 

class MainActivityViewModel(startingTotal : Int) : ViewModel() {
//    private var total = MutableLiveData<Int>()
//    val totalData : LiveData<Int>
//        get() = total

      private val _flowTotal = MutableStateFlow<Int>(0)
      val flowTotal : StateFlow<Int> = _flowTotal
      //get() = _flowTotal



    init {
       // total.value = startingTotal
        _flowTotal.value = startingTotal
    }

    fun setTotal(input:Int){
       // total.value =(total.value)?.plus(input)
        _flowTotal.value = (_flowTotal.value).plus(input)
    }
}

ViewModel 에서 변수를 StateFlow 로 정의하고 Add 버튼을 클릭하면 StateFlow 값을 플러스 하는 함수로 되어있다.

또한 초깃값 0을 가지고 있는 것을 확인 할 수 있다. 

class MainActivity : AppCompatActivity() {
    private lateinit var binding: ActivityMainBinding
    private lateinit var viewModel: MainActivityViewModel
    private lateinit var viewModelFactory: MainActivityViewModelFactory
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        binding = DataBindingUtil.setContentView(this, R.layout.activity_main)
        viewModelFactory = MainActivityViewModelFactory(125)
        viewModel = ViewModelProvider(this,viewModelFactory).get(MainActivityViewModel::class.java)

//        viewModel.totalData.observe(this, Observer {
//            binding.resultTextView.text = it.toString()
//        })

        lifecycleScope.launchWhenCreated {
             viewModel.flowTotal.collect{
                 binding.resultTextView.text = it.toString()
             }
        }

        binding.insertButton.setOnClickListener {
            viewModel.setTotal(binding.inputEditText.text.toString().toInt())
        }
    }
}

액티비티에서는 launchWhenCreate {} 를 통해 뷰 라이프사이클이 Create일때 StateFlow 를 collect 해주었고 값이 바뀔때마다 계속 소비하고 최신값으로 text를 세팅해주게 된다.

 

 

SharedFlow는 StateFlow와 다르게 초깃값(초기상태)를 필요로 하지 않고, 이러한 상태에서 이벤트가 발생했을때 데이터를 emit 할때 사용하면 좋다.

그래서 밑 예제도 버튼을 클릭하면 토스트메시지를 띄워주는 형식으로 사용하였다.

추가로, 좀 더 깊은 구조를 알고 싶으면 다음 포스팅을 참고하면 좋다.

https://myungpyo.medium.com/stateflow-%EC%99%80-sharedflow-32fdb49f9a32

StateFlow 와 SharedFlow

 

뷰모델 코드다.

SharedFlow 는 초깃값을 안가지고 있고 setTtotal() 버튼을 누를떄때 발생하는 이벤트 함수에 값을 emit 하는 코드를 추가했다.

class MainActivityViewModel(startingTotal : Int) : ViewModel() {

      private val _flowTotal = MutableStateFlow<Int>(0)
      val flowTotal : StateFlow<Int> = _flowTotal

	  //SharedFlow는 기본값을 안가진 것을 볼 수 있음
      private val _message = MutableSharedFlow<String>()
      val message : SharedFlow<Int> = _message


    init {
       // total.value = startingTotal
        _flowTotal.value = startingTotal
    }

    fun setTotal(input:Int){
       // total.value =(total.value)?.plus(input)
        _flowTotal.value = (_flowTotal.value).plus(input)
        viewmodelScope.launch{
        	_message.emit("setTotal Success!!!")
        }
    }
}

 

액티비티 코드이다. 뷰모델의 SharedFlow 흐름을 구독 감지하여 토스트 메시지를 띄워준다.

class MainActivity : AppCompatActivity() {
    private lateinit var binding: ActivityMainBinding
    private lateinit var viewModel: MainActivityViewModel
    private lateinit var viewModelFactory: MainActivityViewModelFactory
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        binding = DataBindingUtil.setContentView(this, R.layout.activity_main)
        viewModelFactory = MainActivityViewModelFactory(125)
        viewModel = ViewModelProvider(this,viewModelFactory).get(MainActivityViewModel::class.java)

//        viewModel.totalData.observe(this, Observer {
//            binding.resultTextView.text = it.toString()
//        })

        lifecycleScope.launchWhenCreated {
             viewModel.flowTotal.collect{
                 binding.resultTextView.text = it.toString()
             }
        }

        lifecycleScope.launchWhenCreated {
             viewModel.message.collect{
                 Toast.makeText(this, it, Toast.LENGTH_LONG).show()
             }
             
        }

        binding.insertButton.setOnClickListener {
            viewModel.setTotal(binding.inputEditText.text.toString().toInt())
        }
    }
}

 

 

이 밖에 공식문서를 보니 강의에서 다루지 않는 Cold Stream을 Hot Stream으로 바꿔주는 shareIn 과 같은 것도 있으니 나중에 사용하게되면 추가로 보면 좋을 것 같다.

 

 

 

 

 

댓글과 공감은 큰 힘이 됩니다. 감사합니다!!!!