[RxJava] Combining Operators (결합 연산자) - merge / zip / combineLatest 비교

RxJava의 결합 연산자(combining operator) 중 merge / zip / combineLatest 의 각 특징을 알아보고, 차이점을 비교해 언제 어떤 함수를 써야 할지에 대해 설명한다.

 

목차

 

---

merge()

Flattens two ObservableSources into a single ObservableSource, without any transformation

== 변형 없이 두 개의 ObservableSource를 하나의 ObservableSource로 병합한다

 

특징은 아래와 같다.

• 여러개의 소스에서 나오는 모든 스트림이 그대로 발행된다. 
  • 위의 다이어그램을 보면, merge 함수를 거쳐도 기존 아이템이 고대로 발행된다!
• 하나의 소스에서 에러가 나면 스트림 자체가 종료된다
  • 위의 다이어그램을 보면, 첫 번째 스트림에서 에러가 발생한 이후, 2번째 스트림에서 나온 아이템은 발행되지 않는다!
• 각 소스가 모두 complete 되어야 merge 스트림도 종료된다 (아래 예시에서 확인할 것)

 

코드 예시

Observable.merge(
    observableSource1,
    observableSource2
)
    .subscribe {
        // ...
    }

 

 

 

zip()

combine the emissions of multiple Observables together via a specified function and emit single items for each combination based on the results of this function

== 지정된 함수를 통해 여러 Observable의 아이템을 결합하고 이 함수의 결과에 따라 각 조합에 대해 하나의 아이템을 발행

 

특징은 아래와 같다.

• 각 소스에서 아이템을 하나씩 합쳐서 새로운 아이템을 발행한다.
• 포인트는 각 소스에서 1:1 로 매칭 하여 새로운 아이템을 발행한다는 것.
  • 바로 아래의 zip 그림을 보면, 2번째 스트림에서 C/D 가 연달아 발행되어도, C 에 대응하는 1번째 스트림에서의 3이 발행된 이후에 3C 라는 새로운 아이템이 발행된다!
• 하나의 소스라도 complete 되면 zip 스트림도 종료된다(위 다이어그램과 같이) (merge 와의 차이점)

 

코드 예시

Observable.zip(
    observableSource1,
    observableSource2
) { 
    // 각 스트림에서 발행된 아이템을 조합해 새로운 아이템을 만든다!
    source1, source2 -> "$source1 - $source2" 
}
    .subscribe {
        // ...
    }

 

 

 

 

combineLatest()

when an item is emitted by either of two Observables, combine the latest item emitted by each Observable via a specified function and emit items based on the results of this function

== 두 Observable 중 하나의 소스에서 아이템이 발행되면 지정된 함수를 통해 각 Observable에서 발행된 최신 아이템을 결합하고 이 함수의 결과에 따라 새로운 아이템을 발행

 

특징은 아래와 같다.

• 첫번째 아이템 발행은, 여러 개의 소스에서 각 첫 번째 아이템이 모두 발행되었을 때 발행된다(위 다이어그램에서 1A)
• 그 이후에는 각 소스에서 새로운 아이템이 발행될 때 마다 각 소스별 최신 아이템들이 합쳐져서 새로운 아이템이 발행된다
• merge와 동일하게 각 소스가 모두 complete 되어야 merge 스트림도 종료된다

 

코드 예시

Observable.combineLatest(
    observableSource1,
    observableSource2
) {     
    // 각 스트림에서 발행된 아이템을 조합해 새로운 아이템을 만든다!
    source1, source2 -> "$source1 - $source2" 
}
    .subscribe {
        // ...
    }

 

 

---

 

그래서 언제 어떤 함수를 써야 하냐! (차이점 비교)

	merge	zip	combineLatest
스트림 시작 시점 (첫번째 아이템이 발행되는)	어느 하나의 소스에서 아이템이 발행 됐을 때	각 소스의 첫번째 아이템들이 모두 발행되었을 때	각 소스의 첫번째 아이템들이 모두 발행되었을 때
각 소스별 아이템을 조합해서 새로운 아이템을 만들어내는지	X	O	O
스트림 완료(complete) 시점	각 소스가 모두 complete 되었을 때	하나의 소스라도 complete 되었을 때	각 소스가 모두 complete 되었을 때

 

직접 코드 실행으로 비교해보자

시나리오

아래와 같이 2개의 소스 스트림이 있다고 했을 때, 각 combining operator(결합 연산자)에 따라 발행되는 아이템의 차이를 보자!

 

코드

val source1 = BehaviorSubject.create<Int>()
val source2 = BehaviorSubject.create<String>()

Observable.merge(
    source1,
    source2
)
    .doOnSubscribe {
        print("LOG>> [merge] ")
    }
    .doOnTerminate {
        println("\nLOG>> **** merge Terminate ****")
    }
    .subscribe {
        print("$it - ")
    }

Observable.zip(
    source1,
    source2
) { int, string -> "($int$string)" }
    .doOnSubscribe {
        print("LOG>> [zip] ")
    }
    .doOnTerminate {
        println("\nLOG>> **** zip Terminate ****")
    }
    .subscribe {
        print("$it - ")
    }

Observable.combineLatest(
    source1,
    source2
) { int, string ->
    "($int$string)"
}
    .doOnSubscribe {
        print("LOG>> [combineLatest] ")
    }
    .doOnTerminate {
        println("\nLOG>> **** combineLatest Terminate ****")
    }
    .subscribe {
        print("$it - ")
    }

source1.onNext(1)
source2.onNext("A")
source1.onNext(2)
source2.onNext("B")
source1.onNext(3)
source2.onNext("C")
source2.onNext("D")
source2.onNext("E")
source1.onNext(4)
source1.onNext(5)
source1.onComplete()
source2.onComplete()

 

 

결과

[merge] 1 - A - 2 - B - 3 - C - D - E - 4 - 5 -

[zip] (1A) - (2B) - (3C) - (4D) - (5E) -

[combineLatest] (1A) - (2A) - (2B) - (3B) - (3C) - (3D) - (3E) - (4E) - (5E) -

 

 

 

비교 끝!!!