我正在使用值列表创建一个Observable列表,foreach值是一个自定义Observable.我全部使用merge运行它们,但是我无法检测到哪个触发onNext()或onError()

就像下面的代码：

 List<Observable<MyHttpRsObj>> observables = new ArrayList<>();

    for (String param : paramsList) {
        Observable<MyHttpRsObj> objObservable = MyRestClient.get().doHttpRequest(param);
        observables.add(fileUploadObservable);
    }

    Observable<BaseRs> combinedObservables = Observable.merge(observables);

    combinedObservables.observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(new Subscriber<MyHttpRsObj>() {
                @Override
                public void onCompleted() {
                    //called only once when all Observables finished
                }

                @Override
                public void one rror(Throwable throwable) {
                    //how to know which Observable has error (which param)
                }


                @Override
                public void onNext(MyHttpRsObj myHttpRsObj) {
                    //how to know which Observable has sccess  (which param)
                }
            });

解决方法:

由于将所有可观察值合并为一个,因此无法知道哪个可观察值触发了错误.

最好的选择是为每个可观察对象使用一个观察者.最后一个用于合并的Observable.

像这样：

 List<Observable<MyHttpRsObj>> observables = new ArrayList<>();

    for (String param : paramsList) {
        //change to connectable Observable
        ConnectableObservable<MyHttpRsObj> objObservable = MyRestClient.get()
                 .doHttpRequest(param)
                 .publish();

       //don't forget to connect
        observable.connect();
        observables.add(fileUploadObservable);

        //subscribe for each observable
        objObservable.observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io())
                .subscribe(new Subscriber<MyHttpRsObj>() {
                    @Override
                    public void onCompleted() {
                        //just partial completed
                    }

                    @Override
                    public void one rror(Throwable throwable) {
                        //you can access param from here

                    }


                    @Override
                    public void onNext(MyHttpRsObj myHttpRsObj) {
                        //access onNext here
                        //you can access param from here
                    }
                });
    }

    Observable<BaseRs> combinedObservables = Observable.merge(observables);

    combinedObservables.observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(new Subscriber<MyHttpRsObj>() {
                @Override
                public void onCompleted() {
                    //called only once when all Observables finished
                }

                @Override
                public void one rror(Throwable throwable) {
                    //don't handle error here
                }


                @Override
                public void onNext(MyHttpRsObj myHttpRsObj) {

                }
            });

PS：使用ConnectableObservable避免发射两次