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

2019-11-20 21:28:56 阅读：349 来源： 互联网

标签：rx-java reactive-programming java android

我有执行方法,它会在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 one rror(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,则每个新输入值仅在concatMap发出的最后一个Observable完成时进行订阅.在底层,concatMap为此使用了SerialSubscription.那应该给您想要的订单.

标签：rx-java,reactive-programming,java,android
来源： https://codeday.me/bug/20191120/2046765.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

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