如何在可观察的流中处理前n个项目并以不同的方式处理其余项

Question

例如，

给定一定数量（m）的数字流（m1，m2，m3，m4，m5，m6 …），然后对前n个项应用变换（2 * i）（n可以小于，等于或大于m），对其余项应用另一个变换（3 * i）。和

返回结果：m1 * 2，m2 * 2，m3 * 3，m4 * 3，m5 * 3，m6 * 3 …（这里假设n = 2）。

我试图使用take（n）和skip（n），然后使用concatwith，但是take（n）看起来将删除序列中的其余项，然后使skip（n）不返回任何内容。

Answer 1

你可以分享你M的流，然后合并到一起take()并skip()流，是这样的：

    int m = 10;
    int n = 8;
    Observable<Integer> numbersStream = Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
            .publish();

    Observable<Integer> firstNItemsStream = numbersStream.take(n)
            .map(i -> i * 2);

    Observable<Integer> remainingItemsStream = numbersStream.skip(n)
            .map(i -> i * 3);

    Observable.merge(firstNItemsStream, remainingItemsStream)
            .subscribe(integer -> System.out.println("result = " + integer));
    numbersStream.connect();

编辑：
@AE
Daphne指出，share()它将与第一个订阅者一起开始发射，因此，如果Observable已开始发射项目，则第二个订阅者可能会错过通知，因此在这种情况下，还有其他可能性：
cache() -将答复所有缓存发出的项目并将其回复给每个新订户，但会牺牲取消订阅的能力，因此需要谨慎使用。
reply().refCount()
-将创建Observable的是reply()以前所有项目的每个新用户（类似缓存），但会取消时，从它的最后一个订户退订。

在这两种情况下，都应考虑内存，因为Observable它将在内存中缓存所有发出的项目。

publish() -在不缓存所有先前项目的情况下，另一种可能性是使用publish()create
ConnectableObservable，并connect()在所有必需的订阅者都订阅之后调用它的方法来开始发射，这样将获得同步，并且所有订阅者将正确获得所有通知。