RxJava Cold & Hot Observable

Cold Observable

Cold Observable은 다음과 같은 특징을 가지고 있습니다.

• 일반적인 옵저버 형태를 말합니다.

• 누가 구독해주지 않으면 데이터를 배출해주지 않습니다.

• 일반적인 웹 요청, 데이터베이스 커리 등이 사용되며 내가 요청하면 결과를 받는 과정을 거칩니다.

• 처음부터 발행하는 것을 기본으로 합니다.

위의 특징을 정리하자면 Cold Observable은 생성된 이후 누군가 구독을 요청했을 때 처음부터 데이터를 발행해준다고 보면 됩니다.

이제 Hot Observable을 보겠습니다.

Hot Observable

• 구독자의 존재 여부와 상관없이 데이터를 배출하는 Observable 입니다.

• 여러 구독자에 선택적으로 고려가 가능합니다.

• 구독 시점으로부터 발행하는 값을 받는 걸 기본으로 합니다.

• 마우스 이벤트, 키보드 이벤트, 시스템 이벤트 등이 주로 사용됩니다.

• 멀티캐스팅도 포함됩니다.

특징을 정리하면 Hot Observable은 데이터 발행 시작한 이후 모든 구독자들에게 같은 데이터를 발행 해줍니다.

개념만 보아서는 잘 이해가 되지 않을 수도 있습니다. 그렇기에 코드로 한번 보겠습니다.

Cold Observable Code

Observable에 구독을 요청했을 때 데이터를 발해하기 시작합니다. 즉 누군가 구독을 하지 않으면 데이터를 발행하지 않습니다. 또한 데이터를 처음부터 끝까지 발행하는 것을 보장합니다.

fun main() {
	
    val coldObservable = Observable.interval(1, TimeUnit.SECONDS)
    
    coldObservable.subscribe {
    	println("First Subscriber : $it")
    }
    
    Thread.sleep(2000)
    
    coldObservable.subscribe {
    	println("Second Subscriber : $it")
    }
    
    Thread.sleep(2000)
}
    

다음과 같이 1씩 증가하는 코드가 있습니다. First Subscriber의 경우는 바로 증가하는 것이지만 2초가 있고난뒤 Second Subscriber가 하나씩 증가합니다.

같은 Observable이라는 인스턴스를 사용했음에도 Observer마다 다른 Observable Stream이 생성되게 되어 있습니다. 조금 더 쉽게 설명하면 유튜브와 비슷합니다. 어떤 사용자가 영상을 보다 중간에 종료를 하였습니다. 그 후 다른 사용자가 영상을 누른다고 앞의 사용자가 종료한 부분이 아닌 처음부터 다시 시작하게 됩니다. 이거와 같이

Cold Observable은 누가 구독하든 무조건 데이터 맨 처음부터 전해주는 특징이 있습니다.

Hot Observable Code

Hot Observable의 경우 구독자의 존재 여부 상관없이 데이터를 발행하는 특징을 가지고 있습니다. 데이터 발행이 시작된 이후로 모든 구독자에게 동시에 같은 데이터를 발행합니다. 위의 예시로 비유하자면 first Subscriber가 2가 된 이후 second는 0으로 나오지만 Hot의 경우는 3으로 출력되는 것입니다.

안드로이드를 예로 들면 클릭 이벤트 등의 UI 이벤트에 사용됩니다. Observer가 있든 말든 이벤트는 계속 발생됩니다. Observer 들은 이를 구독한 이후에 발생하는 이벤트를 받아보게 됩니다.

ConnectableObservable

Hot Observable 구현을 위해 ConntectableObservable이라는 특수한 Observable을 사용합니다. 처음보는 publish() 와 connect() 라는 메서드에 집중해보겠습니다.

fun main() {
	// publish()를 붙이는 순간 Hot Observable로 변환됨 
    
    val hotObservable : ConntectableObservable<Long> = Observable.interval(1, TimeUnit.SECONDS).publish()
    
    // connect()가 호출될 때 곧바로 데이터 발행 시작 
    
    hotObservable.connect() 
    
    hotObservable.subscribe {
    	println("First Subscriber : $it")
    }
    
    Thread.sleep(2000)
    hotObservable.subscribe {
    	println("Second Subscriber : $it")
    }
    Thread.sleep(2000)
}   	

결과는 다음과 같이 출력됩니다.

First Subscriber : 0
First Subscriber : 1
First Subscriber : 2
Second Subscriber : 2
First Subscriber : 3
Second Subscriber : 3

어떠한 subscriber가 있든지 없든지 발행하던 데이터를 계속하여 보내줍니다. 실시간 스트리밍 서비스라고 생각하면 될 것같습니다.

ConnectableObservable에 publish()를 붙이는 순간 해당 스트림은 Hot Observable이 됩니다. 그리고 connect()라는 녀석이 호출될 때 누가 구독하고 있든 말든 데이터를 발행을 시작합니다.

references

http://reactivex.io/documentation/operators
https://healthcoding.tistory.com/27
https://velog.io/@haero_kim/RxJava-Cold-Hot-Observable