RXJava：如何在不完成所有可观察项的情况下组合多个可观察项？

Question

我有多个热观测器，它们可能发射也可能不发射项目。因此，我想把可观察项组合起来，然后如果其中任何一个发出结果，就处理结果，但如果其他可观察项在item发出，就应该一起处理。

observable1 = PublishSubject<>()  
observable2 = PublishSubject<>()

observable1.onNext(1)  
observable1.onNext(2)  
observable2.onNext("Test")  
observable1.onNext(3)

(1, null) 
(2, null)
(2, "Test")
(3, "Test")

还有可能在observable1之前发出observable2

CombineLatest最接近我所需要的，但只有当所有可观察到的对象发出至少一个项时才会发出结果。这个有无反应操作符吗？

Answer 1

您可以对每个源使用StartWith来提供初始值，或者对初始值使用BehaviorSubject，然后对这些扩展的可观察项应用CombineLatest。但是，null在RXJava2中是不允许的，所以必须在可观察的元素类型中找到一个中性值。

PublishSubject<Integer> observable1 = PublishSubject.create()  
PublishSubject<String>  observable2 = PublishSubject.create()

Observable.combineLates(
    observable1.startWith(-100000),
    observable2.startWith(""),
    (a, b) -> a + b
)
.subscribe(System.out::println)
;

observable1.onNext(1)  
observable1.onNext(2)  
observable2.onNext("Test")  
observable1.onNext(3)

或

BehaviorSubject<Integer> observable1 = BehaviorSubject.createDefault(-10000)  
BehaviorSubject<String>  observable2 = BehaviorSubject.createDefault("")

Observable.combineLates(
    observable1,
    observable2,
    (a, b) -> a + b
)
.subscribe(System.out::println)
;

observable1.onNext(1)  
observable1.onNext(2)  
observable2.onNext("Test")  
observable1.onNext(3)