rxjava2:使用连接的Completable并在IO线程中观察它们

起初,我知道网络操作不应该从主线程调用。 这就是为什么我在Schedulers.io()上观察Completable的原因!

我试图连接两个完整的。 这两个完整的使用网络,这就是为什么我订阅Schedulers.io()。 如果我使用concatWith(或者和然后)代码失败与NetworkOnMainThreadException。 这里是kotlin代码:

val singleSubject = SingleSubject.create(); completalbe1.concatWith(completable2) .subscribeOn(Schedulers.io()) .observeOn(Schedulers.io()) .subscribe({ singleSubject.onSuccess("ok") }, { error -> Log.e(tag, error.message, error)//here i got exception singleSubject.onError(error) }) return singleSubject 

如果我重写代码没有完整的链接 – 一切都好。 这是工作代码:

 val singleSubject = SingleSubject.create(); completable1 .subscribeOn(Schedulers.io()) .observeOn(Schedulers.io()) .subscribe({ completable2 .subscribeOn(Schedulers.io()) .observeOn(Schedulers.io()) .subscribe({ singleSubject.onSuccess("ok") }, { error -> Log.e(tag, error.message, error) singleSubject.onError(error) }) }, {error -> Log.e(tag, error.message, error) singleSubject.onError(error) }) return singleSubject 

我想知道为什么第一个片段不工作,但第二个呢?

UPD1:这里是stacktrace:

  android.os.NetworkOnMainThreadException at android.os.StrictMode$AndroidBlockGuardPolicy.onNetwork(StrictMode.java:1273) at libcore.io.BlockGuardOs.recvfrom(BlockGuardOs.java:249) at libcore.io.IoBridge.recvfrom(IoBridge.java:549) at java.net.PlainSocketImpl.read(PlainSocketImpl.java:481) at java.net.PlainSocketImpl.access$000(PlainSocketImpl.java:37) at java.net.PlainSocketImpl$PlainSocketInputStream.read(PlainSocketImpl.java:237) at okio.Okio$2.read(Okio.java:139) at okio.AsyncTimeout$2.read(AsyncTimeout.java:237) at okio.RealBufferedSource.exhausted(RealBufferedSource.java:56) at okhttp3.internal.connection.RealConnection.isHealthy(RealConnection.java:498) at okhttp3.internal.connection.StreamAllocation.findHealthyConnection(StreamAllocation.java:133) at okhttp3.internal.connection.StreamAllocation.newStream(StreamAllocation.java:100) at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:42) at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92) at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67) at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:93) at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92) at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67) at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93) at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92) at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:120) at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92) at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67) at okhttp3.logging.HttpLoggingInterceptor.intercept(HttpLoggingInterceptor.java:211) at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92) at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67) at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:185) at okhttp3.RealCall.execute(RealCall.java:69) at retrofit2.OkHttpCall.execute(OkHttpCall.java:180) at com.jakewharton.retrofit2.adapter.rxjava2.CallObservable.subscribeActual(CallObservable.java:41) at io.reactivex.Observable.subscribe(Observable.java:10955) at com.jakewharton.retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34) at io.reactivex.Observable.subscribe(Observable.java:10955) at io.reactivex.internal.operators.observable.ObservableIgnoreElementsCompletable.subscribeActual(ObservableIgnoreElementsCompletable.java:31) at io.reactivex.Completable.subscribe(Completable.java:1664) at io.reactivex.internal.operators.completable.CompletableConcatArray$ConcatInnerObserver.next(CompletableConcatArray.java:89) at io.reactivex.internal.operators.completable.CompletableConcatArray$ConcatInnerObserver.onComplete(CompletableConcatArray.java:65) at io.reactivex.internal.operators.completable.CompletableCreate$Emitter.onComplete(CompletableCreate.java:64) at com.catalyst.opti.AppManager$transferImage$1$subscribe$1.onStateChanged(AppManager.kt:323) at com.amazonaws.mobileconnectors.s3.transferutility.TransferStatusUpdater$1.run(TransferStatusUpdater.java:172) at android.os.Handler.handleCallback(Handler.java:742) at android.os.Handler.dispatchMessage(Handler.java:95) at android.os.Looper.loop(Looper.java:154) at android.app.ActivityThread.main(ActivityThread.java:5527) at java.lang.reflect.Method.invoke(Native Method) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:739) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:629) 

UPD2:

completable1是一个将文件上传到AWS S3的function:

 private fun transferImage(imageName: String, image: File): Completable { return Completable.create(object : CompletableOnSubscribe { override fun subscribe(e: CompletableEmitter) { val transferObserver = transferUtility.upload("some", imageName, image) transferObserver.setTransferListener(object : TransferListener { override fun onProgressChanged(id: Int, bytesCurrent: Long, bytesTotal: Long) { Log.i(tag, "bytesCurrent: $bytesCurrent, bytesTotal: $bytesTotal") } override fun onStateChanged(id: Int, state: TransferState?) { if (state == TransferState.COMPLETED) { e.onComplete() } } override fun onError(id: Int, ex: java.lang.Exception) { Log.d(tag, "error transfer s3: ${ex.message}", ex) e.onError(ex) } }) } }); } 

completable2是retrofit2调用:

 @POST("some") fun verifyLocation(@Header(AUTH_TOKEN_HEADER) authToken: String, @Body verifyLocation: VerifyLocation): Completable 

我猜想transferObserver.setTransferListener调用主线程的回调,然后在主线程上订阅completable2 。 您必须将subscribeOn(Schedulers.io())应用于completable2 ,就像在其他示例中一样。

 val singleSubject = SingleSubject.create(); completalbe1.subscribeOn(Schedulers.io()) .concatWith(completable2.subscribeOn(Schedulers.io())) // <----------------------- .observeOn(Schedulers.io()) .subscribe({ singleSubject.onSuccess("ok") }, { error -> Log.e(tag, error.message, error)//here i got exception singleSubject.onError(error) }) return singleSubject 

subscribeOn会影响订阅(侧面)效果,但是当您在主线程上调用onComplete时,您的completalbe1会有一个观察效果。