将Rx Observable拆分为多个流并单独处理

Question

这是我试图完成的图片.

--abca - BBB - 一个

分裂成

--a ----- a ------- a - >一个流

---- b ------ bbb --- - > b stream

------ c ---------- - > c stream

然后,能够

a.subscribe()
b.subscribe()
c.subscribe()

到目前为止,我发现的所有内容都使用groupBy()拆分流,但随后将所有内容折叠回单个流并在同一个函数中处理它们.我想要做的是以不同的方式处理每个派生流.

我现在正在做的方式是做一堆过滤器.有一个更好的方法吗？

Answer 1

像馅饼一样简单,只需使用 filter

scala中的一个例子

import rx.lang.scala.Observable

val o: Observable[String] = Observable.just("a", "b", "c", "a", "b", "b", "b", "a")
val hotO: Observable[String] = o.share
val aSource: Observable[String] = hotO.filter(x ? x == "a")
val bSource: Observable[String] = hotO.filter(x ? x == "b")
val cSource: Observable[String] = hotO.filter(x ? x == "c")

aSource.subscribe(o ? println("A: " + o), println, () ? println("A Completed"))

bSource.subscribe(o ? println("B: " + o), println, () ? println("B Completed"))

cSource.subscribe(o ? println("C: " + o), println, () ? println("C Completed"))

您只需要确保源可观察性很热.最简单的方法就是share它.

Answer 2

你没有崩溃Observables的groupBy.您可以改为订阅它们.

像这样的东西:

    String[] inputs= {"a", "b", "c", "a", "b", "b", "b", "a"};

    Action1<String> a = s -> System.out.print("-a-");

    Action1<String> b = s -> System.out.print("-b-");

    Action1<String> c = s -> System.out.print("-c-");

    Observable
            .from(inputs)
            .groupBy(s -> s)
            .subscribe((g) -> {
                if ("a".equals(g.getKey())) {
                    g.subscribe(a);
                }

                if ("b".equals(g.getKey())) {
                    g.subscribe(b);
                }

                if ("c".equals(g.getKey())) {
                    g.subscribe(c);
                }
            });

如果语句看起来有点难看,但至少你可以分别处理每个流.也许有一种方法可以避免它们.