java - Observable merge() 检测哪个 observable 被触发

Question

我正在使用值列表创建一个 Observable 列表，每个值都是一个自定义 Observable。我使用合并运行它们，但我无法检测到哪个触发 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 onError(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)
                }
            });

Answer 1

不可能知道哪个可观察值触发了错误，因为您将所有可观察值合并为一个。

最好的办法是为每个可观察对象使用一个观察者。最后一个用于合并的 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 onError(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 onError(Throwable throwable) {
                    //don't handle error here
                }


                @Override
                public void onNext(MyHttpRsObj myHttpRsObj) {

                }
            });

附言:使用ConnectableObservable避免发射两次