서론
이전과 이어서 RxJava 1장에 대해서 정리해보겠습니다. 이번에는 이전에 배웠던 Publisher, Subscriber의 구현인 Flowable, Observable, Subscriber, Observer에 대해서 알아보고 후에 Disposable과 마블 다이어그램, Cold 생성자, Hot 생성자에 대해서 정리해보겠습니다.
1. RxJava의 기본
1.3
기본 구조
이전에 말했던 것처럼 RxJava는 데이터를 만들고 통지하는 생산자와 통지된 데이터를 받아 처리하는 소비자로 구성됩니다. 이 생산자를 소비자가 구독해 생산자가 통지한 데이터를 소비자가 받게됩니다. RxJava에서 크게 두가지로 나뉠 수 있는데, Reactive Stream을 지원하는 Flowable(생산자)와 Subscriber(소비자), Reactive Stream을 지원하지 않고, 배압기능이 없는 Observable(생산자)과 Observer(소비자)가 있습니다.
Flowable은 Publisher를 구현한 클래스고, Subscriber은 Reactive Streams의 클래스입니다. 그래서 기본적인 메커니즘은 같습니다! 생산자인 Flowable로 구독 시작(onSubscribe), 데이터 통지(onNext), 에러 통지(onError), 완료 통지(onComplete)를 하고, 각 통지를 받은 시점에 소비자인 Subscriber로 처리합니다. 그리고 Subscription으로 데이터 개수 요청과 구독 해지를 합니다.
Observable은 Reactive Streams를 구현하지 않아서, Reactive Streams 인터페이스를 사용하지 않습니다. 하지만 기본적인 메커니즘과 구성은 Flowable이랑 거의 같습니다. 생산자인 Observable(생산자)에서 구독 시작(onSubscribe), 데이터 통지(onNext), 에러 통지(onError), 완료 통지(onComplete)를 하면 Observer(소비자)에서 통지를 받습니다. 다만 Observable과 Observer 구성은 통지하는 데이터 개수를 제어하는 배압기능(onSubscribe() 메소드에서 인자로 Subscription을 받아, request에서 데이터를 요청할 때 파라미터로 준 Long 값)이 없기 때문에 데이터 개수를 요청하지 않습니다.
그래서 Subscription 대신 Disposable이라는 구독 해지 메소드가 있는 인터페이스를 이용합니다! 이 Disposable은 onSubscribe() 메소드의 인자로 Observer에 전달되며, 구독 해지를 위한 메소드 dispose()와 isDisposed()가 있습니다! 이때문에 Observable과 Observer 간에 데이터 교환시에는 Flowable과 Subscriber처럼 데이터 개수 요청은 하지 않고, 데이터가 생성되자마자 Observer에 통지됩니다.
연산자
RxJava에서는 생산자가 통지한 데이터가 소비자에게 가기 전, 데이터를 가공하여 필요한 데이터만 전해주거나, 소비자가 사용하기 쉽게 변경해야 할때가 있습니다. 이처럼 데이터를 생성하거나 필터링, 변환하는 메소드를 연산자라고 합니다. 그리고 이러한 연산자들을 연결해나가며 최종 통지 데이터의 단순한 처리만을 도울 수 있습니다. 예제로 Flowable을 이용해 출력해보겠습니다.
// 인자의 데이터를 순서대로 통지하는 Flowable을 생성한다.
val flowable = Flowable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
// 짝수에 해당하는 데이터만 통지한다
.filter { data -> data % 2 == 0 }
// 데이터를 100배로 변환한다.
.map { data -> data * 100 }
flowable.subscribe { data -> println("data= $data") }
=> data= 200
data= 400
data= 600
data= 800
data= 1000이 처럼 데이터를 메소드 체인에 통과시켜 원본 Flowable/Observable이 통지하는 데이터를 최종적으로 어떤 데이터로 통지할 것인지를 제어할 수 있습니다. 이는 연산자를 설정한 시점에서 그 처리가 실행되는 것이 아니라, 통지 처리가 시작되고, 통지를 받는 시점에 설정한 처리가 실행됩니다. 그런데, RxJava에서는 이러한 연산자를 설정한 순서가 실행하는 작업에 영향을 미치기 때문에 주의해서 사용해야 합니다.
또한 RxJava는 대부분 함수형 인터페이스를 인자로 받습니다. 이 함수형 인터페이스의 구현은 함수형 프로그래밍의 원칙에 따라 같은 입력을 받으면 매번 같은 결과를 반환하며, 기본으로 그 입력값과 처리는 외부 객체나 환경에 의해 변화가 일어나야하지 않아야 합니다.
위에서 말한 데이터의 상태를 변경하거나 처리 작업의 외부에서 어떤 변화를 주는 것을 부가 작용(사이드 이펙트)라고 하는데, 데이터를 통지하고, 소비자가 전달 받기까지는 이러한 부가 작용의 발생을 피하는 것이 좋습니다. 이러한 부가 작용의 예시라면 객체의 상태를 변경해 외부에서도 참조가 가능한 객체에 어떤 변화를 주거나, DB의 내용을 변경하는 것과 같은 것입니다. 이러한 변경이 데이터 통지 처리에 영향을 주는것을 피하지 않는다면, 책임 범위가 너무 커져 관리가 어렵습니다. 이러한 사이드 이펙트는 메소드 체인에서가 아닌, 최종 데이터를 받아 처리하는 소비자 측에서 이루어질 수 있도록 합니다.
이러한 사이드 이펙트가 발생하지 않는다면, 여러 쓰레드에서 공유하는 자원이 없어 쓰레드 안전을 보장할 수 있습니다. 이는 싱글 쓰레드에서 처리하던 것을 비동기 처리해도, 로직 자체는 변경되지 않습니다!
비동기 처리
RxJava는 데이터를 통지하는 측, 받는 측의 범위를 분리할 수 있게 설계됐습니다. 즉 서로 다른 쓰레드에서 실행할 수 있습니다. 이는 데이터를 통지하는 측이 무엇을 하더라도 데이터를 받는 측의 처리가 받은 데이터에 의해서만 바뀌게 된다면 비동기로 쉽게 전환할 수 있습니다.
RxJava에서는 개발자가 직접 쓰레드를 관리하지 않게, 쓰레드를 관리하는 스케줄러를 제공해줍니다. 스케줄러는 생산자 부분과 소비자부분에 지정해줄 수 있습니다. 생산자 부분은 Flowable/Observable이 데이터 통지를 어떤 스케줄러에서 처리할지를 제어하며, 소비자는 데이터의 필터나 변환을 하는 메서드와 소비자 등이 데이터의 수신 처리를 어느 스케줄러에서 할지를 제어합니다. 하지만 비동기 처리 또한 주의를 기울여야 합니다. 데이터를 통지하는 측, 받는 측은 데이터 통지 시에만 데이터를 주고 받아야 하며, 그 이외의 요인으로 서로의 행동에 영향을 주지 않아야 합니다.
예를 들어, 생산자와 소비자 사이에서 이루어지는 처리를 외부 변수를 참조해 바꾸면, 비동기 처리에 영향을 줍니다. 아래 예시를 보겠습니다!
enum class State {
ADD, MULTIPLY
}
private lateinit var state: State
fun main() {
state = State.ADD
// 300밀리 초마다 데이터를 통지하는 Flowable 생성
val flowable = Flowable.interval(300L, TimeUnit.MILLISECONDS)
// 7건 까지 통지
.take(7)
// 각 데이터를 계산한다
.scan { sum, data ->
if (state == State.ADD) {
return@scan sum + data
} else {
return@scan sum * data
}
}
// 구독하여 받은 데이터를 출력한다
flowable.subscribe { data -> println("data= $data") }
// 잠시 기다렸다가 곱셈으로 변경
Thread.sleep(1000)
state = State.MULTIPLY
Thread.sleep(2000)
}
=> data= 0
data= 1
data= 3
계산 방법 변경
data= 9
data= 36
data= 180
data= 1080
결과를 보니 중간에 덧셈에서 곱셈으로 변경됐습니다..! 이것이 만약 데이터를 덧셈으로 모두 출력한 뒤 곱셈으로 바꿀 예정이였다면, 의도대로 출력이 되지 않았습니다.
이 예제에서는 직접 처리중 외부 변수를 바꾸고 있으므로 실제로는 이런 일이 없다고 생각할 수도 있으나, 프로그램이 복잡해지고, 비동기 처리중 외부 변수를 참조하면 이런일이 일어날 가능성도 충분합니다.
그래서 비동기로 처리할 때는 생산자에서 소비자까지의 처리가 노출되지 않게 폐쇄적으로 개발하면 이러한 위험을 어느정도 줄일 수 있습니다. 기본적으로 생산자가 외부에서 데이터를 받아 데이터를 생성할 때와 소비자가 받은 데이터를 처리하고 이를 외부에 반영할 때만 외부 데이터를 참조하는 것이 좋습니다.
Cold 생산자, Hot 생산자
생산자에는 2가지 종류가 있는데 Cold 생산자, Hot 생산자가 있습니다. 먼저 Cold 생산자는 1개의 소비자와 구독 관계를 맺고, Hot 생산자는 여러 소비자와 구독관계를 갖습니다. 이는 Cold 생산자가 통지하는 데이터의 타임라인은 구독할 때마다 생성되지만, Hot 생산자는 이미 생성한 통지 데이터의 타임라인에 나중에 소비자가 참가하는 것을 허용합니다.
그래서 Cold 생산자를 구독하면 생산자 처리가 시작되지만, Hot 생산자는 구독해도 생산자 처리가 시작되지 않을 수 있습니다. 또한 Hot 생산자는 이미 처리를 시작한 생산자를 구독한다면 구독한 시점부터 데이터를 받게 되고, 같은 데이터를 여러 소비자가 받을 수 있습니다.
만약 1부터 10까지 데이터를 방출하는 생산자가 있을 때, Hot 생산자를 이용하고 있다면, 3을 방출할 때 구독을 하게되면, 3부터 데이터를 받게 됩니다.
이제부터는 Hot, Cold 생산자의 종류에 대해서 알아보겠습니다! (기본적으로 생산되는 생산자는 Cold 생산자 입니다)
ConnectableFlowable / ConnectableObservable
이 두가지는 Hot Flowable / Observable이며 여러 Subscriber, Observer에서 동시에 구독할 수 있습니다. 또한 Cold와 달리 subscribe 메소드를 호출해도 처리를 시작하지 않고 connect 메소드를 호출해야만 처리를 시작합니다. 이를 이용해 여러 구독자를 구독하게한 뒤 동시에 데이터를 통지할 수 있습니다. 또한 처리를 시작하는 connect 외에도 몇가지 메소드가 존재하며, 끝에 Flowable / Observable을 생성해줍니다(처리 후 변환해주기 위해 사용!).
refCount
위에서 말한 변환 함수입니다. refCount는 ConnectableFlowable / ConnectableObservable에서 Flowable / Observable을 생성합니다. 이 Flowable / Observable을 다른 소비자가 구독중이라면 도중에 구독하더라도 같은 타임라인에서 생성되는 데이터를 통지합니다.
refCount 메소드에서 생성한 Flowable / Observable은 더 이상 ConnectableFlowable / ConnectableObservable이 아니기 때문에 connect 메소드가 없습니다. 그러므로 아직 구독 상태가 아니라면 subscribe 메소드가 호출될 때 처리를 시작합니다. 또한 refCount로 생성한 Flowable / Observable은 처리 완료 후 구독이 모두 해지된 뒤 다시 subscribe 메소드가 호출되면 새로운 처리를 다시 시작합니다!
autoConnect
autoConnect 메소드는 ConnectableFlowable / ConnectableObservable에서 지정한 개수의 구독이 시작된 시점에서 처리를 시작하는 Flowable/ Observable을 생성합니다. autoConnect의 인자가 없다면 처음 subscribe 메소드가 호출된 시점에 처리를 시작하고, 인자로 구독 개수를 지정해주면 개수에 도달한 시점에서 처리를 시작합니다. autoConnect를 이용해 만든 Flowable/Observable은 처리가 완료된 뒤 또는 모든 구독이 해지된 뒤 다시 subscribe를 호출해도 처리가 다시 시작되지 않습니다.
처음부터 Hot 생산자가 아닌, Cold 생산자를 Hot 생산자로 바꿔주는 연산자도 존재합니다!
(여기부터는 사용해보진 못했고 이론으로만 알고 있습니다!)
publish
publish 메소드는 Cold인 Flowable / Observable에서 ConnectableFlowable / ConnectableObservable을 생성하는 연산자 입니다. 이 메소드로 생성한 ConnectableFlowable / ConnectableObservable은 처리를 시작한 뒤 구독하면, 구독한 이후에 생성된 데이터부터 새로운 소비자에게 통지합니다.
replay
replay 메소드는 publish와 비슷하게 Cold인 Flowable / Observable에서 ConnectableFlowable / ConnectableObservable을 생성하는 연산자 입니다. 이 메소드로 생성한 ConnectableFlowable / ConnectableObservable은 통지한 데이터를 캐시하고, 처리를 시작한 뒤 구독하면 캐시한 데이터를 먼저 새로 구독한 소비자에게 통지, 후에 같은 데이터를 통지합니다. 인자가 없이 사용하면 모든 데이터를 캐시, 인자로 지정한 시간동안 지정한 개수만큼 데이터를 캐시합니다.
share
share 메소드는 여러 소비자가 구독할 수 있는 Flowable/Observable을 생성합니다. 이 share은 다른 메소드와는 다르게 ConnectableFlowable / ConnectableObservable을 생성하지 않고 같은 타임라인을 유지하는 생산자를 만듭니다. 실질적으로 flowable.publish.refCount()와 동일합니다!
1.4
마블 다이어그램
Rxjava 코드 내부를 보거나 ReactiveX 문서를 보다 보면 마블 다이어그램을 자주 접할 수 있습니다.
다음은 Reactivex 공식 홈페이지에 올라와있는 마블 다이어그램입니다.
filter에 대한 마블 다이어그램입니다!
먼저 가장 윗줄에 있는 도형과 선에 대해서 이야기해보면, 선은 통지되는 원본 데이터의 타임라인입니다. 도형들은 통지되는 데이터이고, 아래로 향하는 점선은 데이터가 변환되는 흐름을 나타냅니다. 길쭉한 네모는 메소드를 표시하고 있고, 아래의 선과 도형은 원본 데이터가 메소드를 통과한 뒤 결과로 통지되는 데이터와 타임라인 입니다. 가장 끝의 세로선은 처리가 정상 종료됐음을 의미하며, 하단의 세로선은 결과 Flowable/Observable이 완료 통지함을 나타냅니다.
그림의 filter 메소드는 지정한 조건에 맞는 데이터만 결과 데이터로 통지합니다. 또한 X표시는 에러가 발생해 처리가 종료됨을 나타냅니다.
다음은 zip 메소드의 마블 다이어그램입니다!
모양과 색을 나타내는 타임라인이 있고, zip함수를 통과하여 두 데이터를 받아 새로운 데이터를 생성하는 메소드입니다.
이처럼 마블 다이어그램은 시간 경과에 따라 어떤 결과가 발생하는지를 예로 보여주기 때문에 많은 문서에서 사용합니다. 문장만으로는 이해가 안가더라도, 마블 다이어그램을 통해 쉽게 이해가 가능합니다. 이러한 마블 다이어그램에 익숙해지면 머릿속으로 마블 다이어그램을 그려보기를 추천합니다!!
1.5
RxJava 예제 실습
이제 RxJava를 실습해보겠습니다. (이번 포스트는 실습까지 하고 다음 포스트에 1장 마무리입니다!)
먼저 build.gradle에 RxJava를 추가해줍니다. (책은 RxJava 2.x 버전이지만 그냥 최신 버전으로 추가했으며, 실습을 Kotlin으로 바꾸어 했기 때문에, rxkotlin을 implementation 했습니다.)
dependencies {
implementation "org.jetbrains.kotlin:kotlin-stdlib"
testImplementation 'org.junit.jupiter:junit-jupiter-api:5.6.0'
testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
implementation "io.reactivex.rxjava3:rxkotlin:3.0.1"
}Flowable 실습
fun main() {
val flowable = Flowable.create<String>(FlowableOnSubscribe {
val datas = listOf("Hello,World","안녕 rxJava")
for(data in datas){
if(it.isCancelled){
// 구독 해지시 처리 중단
return@FlowableOnSubscribe
}
// 데이터 통지
it.onNext(data)
}
// 완료 통지
it.onComplete()
},BackpressureStrategy.BUFFER)
flowable
// Subscriber 처리를 개별 스레드에서 진행
.observeOn(Schedulers.computation())
// 구독한다
.subscribe(object : Subscriber<String>{
private lateinit var subscription: Subscription
//구독 시작시 처리
override fun onSubscribe(s: Subscription?) {
if (s != null) {
subscription = s
}
// 받을 데이터 개수를 요청한다
subscription.request(1L)
}
// 데이터를 받을 때 처리
override fun onNext(t: String?) {
val threadName = Thread.currentThread().name
println("$threadName : $t")
subscription.request(1L)
}
// 에러 통지시 처리
override fun onError(t: Throwable?) {
t?.printStackTrace()
}
// 완료 통지시 처리
override fun onComplete() {
val threadName = Thread.currentThread().name
println("$threadName : 완료")
}
})
Thread.sleep(1000L)
}
=> RxComputationThreadPool-1 : Hello,World
RxComputationThreadPool-1 : 안녕 rxJava
RxComputationThreadPool-1 : 완료실제 데이터 통지하는 처리는 create메소드의 첫 번째 인자인 FlowableOnSubscribe 인터페이스의 subscribe 메소드에서 이루어집니다. 그리고 subscribe 메소드의 인자인 FlowableEmitter 인터페이스를 통해 Subscriber에 통지합니다. (사실 한칸 더 들어가서 Emitter를 상속하고 있는데, Emitter의 인터페이스를 구현합니다)
내부에서 구독 해지됐다면, 종료되도록 합니다. 또한 모든 데이터를 통지한 뒤에 onComplete메소드를 호출해 Subscriber에 Flowable의 처리가 끝났다는 것을 통지합니다. 이 메소드도 구독을 해지하면 통지가 이루어지지 않습니다. 그리고 완료를 통지한 뒤, 어떤 통지도 하면 안된다는 규약이 있으니, omComplete를 호출한 뒤 아무것도 하지 않게 구현하는 것이 중요합니다.
그리고 마지막 create 메소드의 인자에 BackpressureStarategy를 지정합니다. 이 예제에서는 BUFFER을 지정하는데, 이 옵션은 Flowable의 데이터 생성 속도가 Subscriber의 데이터 처리 속도보다 빠를 때 통지하지 못하고 대기하는 데이터를 모두 버퍼링해 통지할 수 있을 때까지 유지하게 하는 옵션입니다.(옵션이 5가지 있으니 모두 찾아보시길 바랍니다! - BackpressureStartegy로 찾아보시면 됩니다.)
이제 소비자 부분에서 Subscriber의 인터페이스 구현을 살펴보면, onSubscribe에서 request(1L) 해주고 있습니다. 주의사항은 onSubscribe 메소드 내에서 데이터 개수를 요청하지 않으면 Flowable이 통지를 시작할 수 없기때문에 주의해야 합니다.
onNext부분에서 request로 데이터를 요청하지 않으면 처음에 요청한 수만큼 데이터를 통지하고, 데이터가 통지되지 않으니 주의해야 합니다.
또한 추가적으로 Subscriber는 자신이 처리할 수 있는 속도로 데이터를 받게 Flowable이 통지하는 데이터 개수를 제한할 수 있습니다. 하지만 데이터 개수를 요청하고 받은 데이터를 처리하는 일을 반복하다 보면 어떠한 문제가 발생해 데이터 개수를 요청하지 못하는 문제가 발생할 수도 있습니다. 이렇게 된다면 Subscriber가 데이터를 받을 수 잇는 상황이어도 Flowable이 데이터 요청을 대기하는 상태가 지속돼 처리가 멈출 수도 있습니다.
또한, Flowable과 Subscriber의 처리속도가 비슷하거나 통지하는 데이터 개수가 많지 않으면 통지하는 데이터 개수를 제한할 필요가 없을 때도 많습니다. 이때 onSubscribe 메소드에서 실행하는 request 메소드에 Long.MAX_VALUE를 설정해주면 생산자는 데이터 개수의 제한없이 계속해서 데이터를 통지해 이런 위험을 줄일 수 있습니다!
override fun onSubscribe(s: Subscription?) {
if (s != null) {
subscription = s
}
// 받을 데이터 개수를 요청한다
subscription.request(Long.MAX_VALUE)
}
// 데이터를 받을 때 처리
override fun onNext(t: String?) {
val threadName = Thread.currentThread().name
println("$threadName : $t")
}
위의 예제에서는 onSubscribe에서 Long.MAX_VALUE로 설정해주어 onNext에 request가 없습니다!
또한 중간에 구독을 해지해야한다면, Subscription의 cancel 메소드를 호출합니다!
observeOn 메소드
RxJava에서 데이터를 통지하는 측과 전달받는 측의 처리를 각각 별도의 쓰레드에서 실행할 때 쓰레드를 관리하는 Scheduler 객체를 observeOn 메소드의 인자로 설정해 데이터를 받는 측의 처리를 어떤 쓰레드에서 실행할지를 지정할 수 있습니다.
같은 쓰레드에서 데이털르 통지하는 측과 처리하는 측이 실행되면 Flowable이 데이터를 통지한 뒤에 이 데이터를 전달받은 Subscriber의 처리가 끝날 때까지 Flowable은 다음 데이터 통지를 미루고 기다리게 됩니다.
하지만 다른 쓰레드를 이용한다면 통지하는 측은 받는 측의 처리를 기다리지 않고 다음 작업을 할수 있습니다.
Observable 실습
Observable은 Flowable과 비슷해 차이점만 기술하겠습니다!
fun main() {
val observable = Observable.create<String> {
val datas = listOf("Hello,world", "안녕 ,RxJava")
for (data in datas) {
if (it.isDisposed) {
return@create
}
it.onNext(data)
}
it.onComplete()
}
observable.observeOn(Schedulers.computation())
.subscribe(object : Observer<String> {
override fun onSubscribe(d: Disposable) {
// 아무것도 안한다!
}
override fun onNext(t: String) {
println("${Thread.currentThread().name} : $t")
}
override fun onError(e: Throwable) {
e.printStackTrace()
}
override fun onComplete() {
println("${Thread.currentThread().name} : 완료")
}
})
Thread.sleep(500L)
}
=> RxComputationThreadPool-1 : Hello,world
RxComputationThreadPool-1 : 안녕 ,RxJava
RxComputationThreadPool-1 : 완료Flowable과 매우 흡사하지만 몇 가지 차이점이 있습니다! 먼저 onSubscribe에서 아무런 처리도 해주지 않습니다.(물론 중간에 구독을 취소(dispose)하고 싶다면 이전처럼 onSubscribe에서 Disposable을 내부에 가지고 있을 수 있도록 해주어야 합니다!)
그리고 Flowable은 onSubscribe에서 Subscription 인터페이스를 가지고 있었지만, Observable은 Disposable을 가지고 있습니다.
또한 Observable은 배압기능이 없기때문에 request를 해줄 필요 없이 데이터를 받을 때만 처리해줍니다.
그래서 Flowable, Observable 뭘 써야되는데...?
이 책에서는 이럴 때 Flowable, Observable을 쓰라고 합니다!
Flowable
- 대량 데이터 (10,000건 이상)를 처리할 때
- 네트워크 통신이나 데이터베이스 등의 I/O 처리를 할 때
Observable
- 소량의 데이터 (1,000건 이하)를 처리할 때
- GUI 이벤트
- 데이터 처리가 기본으로 동기 방식이며, 자바 표준의 Stream 대신 사용할 때
또한 일반적으로 Observable이 Flowable보다 오버헤드가 적다고 알려져 있으니, 성능이 중요하다면 고려해볼만한 점입니다.
또한 RxJava가 서버에서 사용되는지, 모바일에서 쓰이는지 또한 고려해야하는 점중 하나이고, 서버와 클라이언트 간 연결을 맺어야 한다면 접속하는 클라이언트 수 또한 고려해야 합니다.
이러한 기준으로(사실 더 많고 복잡합니다!) 잘 판단하여 사용하는 것이 중요합니다.
그런데 개인적으로 안드로이드에서는 Observable이 더 많이 쓰이는 것 같습니다...(개인적인 생각입니다) GUI 이벤트라던지, 대량의 데이터를 빠르게 처리해야할 요소가 적기 때문인 듯합니다...!!
마무리
이번 포스트는 여기까지이고, 다음 포스트에서 마지막으로 RxJava의 전체적인 구성에 대해서 알아보겠습니다. 감사합니다.
