RxJava2发布

Question

有什么区别

ObservableTransformer {
    Observable.merge(
        it.ofType(x).compose(transformerherex),
        it.ofType(y).compose(transformerherey)
    )
}

和

ObservableTransformer {
    it.publish{ shared ->
        Observable.merge(
            shared.ofType(x).compose(transformerherex),
            shared.ofType(y).compose(transformerherey)
        )
    }
}

当我使用这两个运行我的代码时,我得到了相同的结果.发布做了什么.

Answer 1

不同之处在于,顶级变压器将从下游为单个订阅订阅上游两次,复制通常不需要的上游的任何副作用:

Observable<Object> mixedSource = Observable.<Object>just("a", 1, "b", 2, "c", 3)
      .doOnSubscribe(s -> System.out.println("Subscribed!"));


mixedSource.compose(f ->
   Observable.merge(
      f.ofType(Integer.class).compose(g -> g.map(v -> v + 1)),
      f.ofType(String.class).compose(g -> g.map(v -> v.toUpperCase()))
   )
)
.subscribe(System.out::println);

将打印

Subscribed!
2
3
4
Subscribed!
A
B
C

这里表示的副作用是打印输出Subscribed!取决于实际来源中的实际工作,这可能意味着发送两次电子邮件,两次检索表的行.通过此特定示例,您可以看到即使源类型在其类型中交错,输出也会单独包含它们.

相反,publish(Function)将为每个最终订户建立一个订阅源,因此源上的任何副作用只发生一次.

mixedSource.publish(f ->
   Observable.merge(
      f.ofType(Integer.class).compose(g -> g.map(v -> v + 1)),
      f.ofType(String.class).compose(g -> g.map(v -> v.toUpperCase()))
   )
)
.subscribe(System.out::println);

打印

Subscribed!
A
2
B
3
C
4

因为源是订阅一次,每个项目被多播到两个"武器" .ofType().compose().