如何在同时发出的两个事件流中进行划分和处理？

Question

有一个随机数流。

rx.Observable
.range (0, 1000)
.map (() -> 200d * Math.random ())

我需要说明的是，流程被分成了两部分。数字小于100且大于100。

之后，对于小于100的数字( chain1 )：我需要对网络执行request1，等待应答，然后继续其他运营商的chain1过程。

对于大于100的数字(chain2)：我必须发送另一个request2，等待应答，然后继续执行过程链运算符。

request1和request2不会互相等待，链是并行执行的。但是在链中，处理必须等待请求的响应。

如何做到这一点？

Answer 1

rx.Observable
                        .create(subscriber -> {
                            for (int i = 0; i < 100; i++) {
                                subscriber.onNext(i);
                                Log.i("Iniop", "Create thread name: " + Thread.currentThread().getName());
                            }
                            subscriber.onCompleted();
                        })
                        .onBackpressureBuffer()
                        .observeOn(Schedulers.computation())
                        .subscribeOn(Schedulers.computation())
                        .map(v -> {
                            Log.i("Iniop", "Map thread name: " + Thread.currentThread().getName());
                            return 200d * Math.random();
                        })
                        .groupBy(k -> {
                                    Log.i("Iniop", "Group thread name: " + Thread.currentThread().getName());
                                    return k > 100 ? "yes" : "no";
                                }
                                , v -> v)
                        .forEach(gO -> gO.observeOn(Schedulers.newThread())
                                        .map(v -> new Pair<String, Double>(gO.getKey(), v))
                                        .subscribe(v -> {
                                                    Log.i("Iniop", "Key: " + v.first);
                                                    Log.i("Iniop", "Value: " + v.second);
                                                    Log.i("Iniop", "Thread name: " + Thread.currentThread().getName());
                                                }
                                                , e -> Log.e("Iniop", "Err", e))
                                , e -> Log.e("Iniop", "Err", e));