rx-java2 - 为什么我会收到 NetworkOnMainThreadException？

Question

我已经尝试了一切，但没有得到任何结果。我只知道 subscribeOn 方法将操作切换到后台线程并防止 UI 线程阻塞，我应该期望后台线程上发生的任何事情都不会影响用户流程。现在我不确定我是否得到了有关 subscribeOn() 方法的错误信息，还是我的代码中有错误？

String profileUrl = BuildConfig.BASEURL + "pharma/patient/profile?username=" + caregiverNo;

    try {
        Observable.just(OkhttpClientManager.getInstance(APP_NAME, APP_VERSION).getRequest(profileUrl))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        callback.getDisposable(d);
                    }

                    @Override
                    public void onNext(String s) {
                        CaregiverUserResponse response = new Gson().fromJson(s, CaregiverUserResponse.class);
                        if (response.getResult().equals(Constants.API_CALL_SUCCESS)) {
                            updateCaregiverCache(response);
                            getUserData(callback, patientNo, authToken);
                        } else {
                            callback.onError(new NetworkError(new Throwable(response.getError())));
                        }
                    }

                    @Override
                    public void onError(Throwable e) {
                        callback.onError(new NetworkError(e));
                    }

                    @Override
                    public void onComplete() {

                    }
                });
    } catch (Exception e) {
        Log.e(TAG, "getCaregiverProfile: ", e);
    }

这是我的代码，其中getRequest()函数将响应作为字符串返回。我收到此错误NetworkOnMainThreadException并且我知道subscribeOn(Schedulers.io())在后台线程上执行操作。

错误：

android.os.NetworkOnMainThreadException
    at android.os.StrictMode$AndroidBlockGuardPolicy.onNetwork(StrictMode.java:1565)
    at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:389)
    at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:230)
    at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:212)
    at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:436)
    at java.net.Socket.connect(Socket.java:621)
    at okhttp3.internal.platform.AndroidPlatform.connectSocket(AndroidPlatform.java:73)
    at okhttp3.internal.connection.RealConnection.connectSocket(RealConnection.java:246)
    at okhttp3.internal.connection.RealConnection.connect(RealConnection.java:166)
    at okhttp3.internal.connection.StreamAllocation.findConnection(StreamAllocation.java:257)
    at okhttp3.internal.connection.StreamAllocation.findHealthyConnection(StreamAllocation.java:135)
    at okhttp3.internal.connection.StreamAllocation.newStream(StreamAllocation.java:114)
    at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:42)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)
    at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:93)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)
    at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
    at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:126)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)
    at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:254)
    at okhttp3.RealCall.execute(RealCall.java:92)
    at com.healthapp.myadhero.network.OkhttpClientManager.getRequest(OkhttpClientManager.java:64)
    at com.healthapp.myadhero.network.Service.getCaregiverProfile(Service.java:81)
    at com.healthapp.myadhero.network.Service.access$200(Service.java:68)
    at com.healthapp.myadhero.network.Service$2.onNext(Service.java:154)
    at com.healthapp.myadhero.network.Service$2.onNext(Service.java:131)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:201)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:255)
    at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:124)
    at android.os.Handler.handleCallback(Handler.java:883)
    at android.os.Handler.dispatchMessage(Handler.java:100)
    at android.os.Looper.loop(Looper.java:214)
    at android.app.ActivityThread.main(ActivityThread.java:7682)
    at java.lang.reflect.Method.invoke(Native Method)
    at com.android.internal.os.RuntimeInit$MethodAndArgsCaller.run(RuntimeInit.java:516)
    at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:950)

提前致谢。

标签： rx-java2

Answer 1

您getRequest甚至在涉及 RxJava 之前调用它，然后将结果值交给 RxJava。使用just近网络呼叫通常是错误的，您应该使用fromCallable.

String profileUrl = BuildConfig.BASEURL + "pharma/patient/profile?username=" + caregiverNo;

Observable.fromCallable(() -> 
    OkhttpClientManager
    .getInstance(APP_NAME, APP_VERSION)
    .getRequest(profileUrl)
)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<String>() {
     @Override
     public void onSubscribe(Disposable d) {
         callback.getDisposable(d);
     }

     @Override
     public void onNext(String s) {
         CaregiverUserResponse response = new Gson().fromJson(s, CaregiverUserResponse.class);
         if (response.getResult().equals(Constants.API_CALL_SUCCESS)) {
             updateCaregiverCache(response);
             getUserData(callback, patientNo, authToken);
         } else {
             callback.onError(new NetworkError(new Throwable(response.getError())));
         }
     }

     @Override
     public void onError(Throwable e) {
         callback.onError(new NetworkError(e));
     }

     @Override
     public void onComplete() {
     }
 });