java - RxJava Backpressure(快速生产者缓慢的消费者)

我有执行方法，在IO线程上做了一些耗时的网络调用

例

/**
 * network call
 * @param value
 * @return
 */
private Observable<Integer> execute(final int value) {
    return Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {

            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("done " + value);
            subscriber.onNext(value);
            subscriber.onCompleted();
        }
    }).subscribeOn(Schedulers.io());
}

那么我有必须按顺序执行的“命令”列表。 (相继)

示例(Observable.range(x，y)表示命令列表)

public List<Integer> testObservableBackpressure(){
   return Observable.range(0,5).flatMap(new Func1<Integer, Observable<Integer>>() {
        @Override
        public Observable<Integer> call(Integer integer) {
            System.out.println("started " + integer);
            return exeute(integer);
        }
    }).toList().toBlocking().single();
}

这样输出是

started 0
started 1
started 2
started 3
started 4
done 0
done 1
done 2
done 4
done 3

产品比消耗更快

我想要这样的结果

started 0
done 0
started 1
done 1
started 2
done 2
...

但..

public List<Integer> testObservableBackpressure(){

    return Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(final Subscriber<? super Integer> subscriber) {
            Observable.range(0,5).subscribe(new Subscriber<Integer>() {

                @Override
                public void onStart() {
                    request(1);
                }

                @Override
                public void onCompleted() {
                    subscriber.onCompleted();
                }

                @Override
                public void onError(Throwable e) {
                    subscriber.onError(e);
                }

                @Override
                public void onNext(Integer integer) {
                    System.out.println("started " + integer);
                    execute(integer).subscribe(new Action1<Integer>() {
                        @Override
                        public void call(Integer integer) {
                            subscriber.onNext(integer);
                            request(1);
                        }
                    });
                }
            });
        }
    }).toList().toBlocking().single();
}

这样结果如预期

started 0
done 0
started 1
done 1
started 2
done 2
started 3
done 3
started 4

我的问题是是否还有另一种更优雅的方式来解决此问题？

最佳答案

我不确定这里是否需要任何特定的背压策略。只需使用concatMap即可。

如果您使用concatMap而不是flatMap，则每个新输入值仅在Observable发出的最后一个concatMap完成时才订阅。在幕后，concatMap为此使用了SerialSubscription。那应该给你想要的订单。