android - 为什么 Observable 没有在正确的线程上创建？

Question

Observable observable = Observable.from(backToArray(downloadWebPage("URL")))
            .map(new Func1<String[], Pair<String[], String[]>>() {
                @Override
                public Pair<String[], String[]> call(String[] of) {
                    return new Pair<>(of,
                            backToArray(downloadWebPage("URL" + of[0])).get(0));
                }
            });

    observable.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe(
            (new Observer<Pair>() {

                @Override
                public void onCompleted() {
                    // Update user interface if needed
                }

                @Override
                public void onError(Throwable t) {
                    // Update user interface to handle error
                }

                @Override
                public void onNext(Pair p) {
                   offices.add(new Office((String[]) p.first, (String[]) p.second));
                }
           }));

这会运行并且我得到 android.os.NetworkOnMainThreadException。我希望它运行由 subscribeOn() 方法设置的新线程。

Answer 1

假设实际网络请求发生在downloadWebPage() ，错误在您的代码的第一行:

Observable observable = Observable.from(backToArray(downloadWebPage("http://api.ataxcloudapp.com/v1/franchise/listing/?location=" + ZIPCode)))

这相当于:

String[] response = downloadWebPage("http://api.ataxcloudapp.com/v1/franchise/listing/?location=" + ZIPCode)

Observable observable = Observable.from(backToArray(response))

这应该清楚 downloadWebPage被执行 - 在主线程上 - 在任何 Observable 之前甚至被创建，更不用说订阅了。 RxJava 在这方面无法改变 Java 的语义。

然而你可以做的是这样的(未经测试，但应该是正确的):

Observable observable = Observable.create(new Observable.OnSubscribe<String[]>() {
        @Override
        public void call(final Subscriber<? super String[]> subscriber) {
            final String[] response = downloadWebPage("http://api.ataxcloudapp.com/v1/franchise/listing/?location=" + ZIPCode);
            if (! subscriber.isUnsubscribed()) {
                subscriber.onNext(backToArray(response));
                subscriber.onCompleted();
            }
        }
)

现在您的网络请求只会发生在 Observable 之后已订阅，并将移至您在 subscribeOn() 中指定的线程.