のRxのjava - 困難な場合

Question

と仮定

Observable<Integer> obs = Observable.just(1, 2, 3, 4, 5);

私はobsでさえ数字の数、およびobsの各奇数数を乗じobsの各さえ数シーケンスを、必要としますに奇数の数を掛けての数値をobsに掛けます。

I.e.与えられた場合、2つのevensと3つのoddsがあるので、結果シーケンスは

3 (1 * 3) 
4 (2 * 2) 
9 (3 * 3) 
8 (4 * 2) 
15 (5 * 3) 

でなければなりません。

Answer 1

あなたは、各ストリームを反復処理し、各分割ストリームの数を掛けた後、奇数にObservableを分割してそれを達成しても、ストリームすることができます：あなたは、インメモリ上でそれをやっている場合

Observable<Integer> obs = Observable.just(1, 2, 3, 4, 5); 
Observable<Integer> oddsObservable = obs.filter(integer -> integer % 2 == 1); 
Observable<Integer> evenObservable = obs.filter(integer -> integer % 2 == 0); 
Observable.merge(
     oddsObservable.flatMap(odd -> oddsObservable.count().map(count -> odd * count)), 
     evenObservable.flatMap(even -> evenObservable.count().map(count -> even * count)) 
) 
     .sorted() 
     .subscribe(System.out::println); 

あなたがJava 8 Stream APIを使用して、同じことをやって検討するかもしれないリスト/配列は、コードが非常に似ています。私のマシンで

List<Integer> list = Arrays.asList(1, 2, 3, 4, 5); 
List<Integer> oddsList = list.stream().filter(integer -> integer % 2 == 1).collect(Collectors.toList()); 
List<Integer> evenList = list.stream().filter(integer -> integer % 2 == 0).collect(Collectors.toList()); 

Stream.of(oddsList.stream().map(odd -> odd * (int) oddsList.stream().count()), 
    evenList.stream().map(even -> even * (int) evenList.stream().count())) 
    .flatMap(integerStream -> integerStream) 
    .sorted() 
    .forEach(System.out::println); 

は、それは大きさ（〜X12）の順で速く走ります。

時間プロデューサー以上のアイテム/番号IDのソースはObservables（リプレイ/キャッシュなど）への複数回の再サブスクライブを避けるためにリアクティブなアプローチを最適化を検討している場合

Answer 2

Observable<Integer> obs = Observable.just(1, 2, 3, 4, 5); 
List<Integer> count = obs.groupBy(i -> i % 2) 
    .sorted((g1, g2) -> g1.getKey() - g2.getKey()) 
    .concatMap(go -> go.count()) 
    .toList() 
    .toBlocking() 
    .first(); 
obs.map(i -> i * count.get(i % 2)) 
    .forEach(System.out::println); 

トラバース二回。