java - Android 上 .flatMap() 中的奇怪调度程序行为

Question

我一直在尝试确定为什么 .flatMap() 运算符中的以下代码显示为在主线程上运行:

public Observable<Void> getObservable() {                                                                        
    return jobServiceObservable                                                    
            .flatMap(jobService -> {                                                                             
                if (Looper.myLooper() == Looper.getMainLooper()) {                                               
                    Log.d("LooperTest", "On main thread.");                                                      
                } else {                                                                                         
                    Log.d("LooperTest", "Not on main thread.");                                                  
                }                                                                                                

                return jobService.syncReservations(accountUtil.getCurrentAccount());
            })                                                                                                   
            .subscribeOn(Schedulers.io()).observeOn(foregroundScheduler);                                                                                                                                                      
}

如您所见，.subscribeOn() 是通过 Schedulers.io() 调用的，但是日志语句显示 中的代码。 flatMap() 在主线程上运行:

LooperTest: On main thread.

作为完整性检查，我向这段代码的各个部分添加了对 .subscribeOn(Schedulers.io()) 的额外调用:

public Observable<Void> getObservable() {                                                                        
    return jobServiceObservable.subscribeOn(Schedulers.io())                                                     
            .flatMap(jobService -> {                                                                             
                if (Looper.myLooper() == Looper.getMainLooper()) {                                               
                    Log.d("LooperTest", "On main thread.");                                                      
                } else {                                                                                         
                    Log.d("LooperTest", "Not on main thread.");                                                  
                }                                                                                                

                return jobService.syncReservations(accountUtil.getCurrentAccount()).subscribeOn(Schedulers.io());
            })                                                                                                   
            .subscribeOn(Schedulers.io()).observeOn(foregroundScheduler);                                                                                                                                                         
}

但是，日志语句似乎显示相同的结果。接下来，在不更改任何代码的情况下，我清理了构建并重新启动了我的模拟器。下一次运行时，打印了以下内容:

LooperTest: Not on main thread.

这很奇怪，因为没有进行任何代码更改。同样，在不更改代码的情况下，我清理了构建并重新启动了模拟器。在下一次运行中，打印了以下内容:

LooperTest: On main thread.

我再次清理构建，然后关闭并打开一个不同类型的新模拟器。运行后打印如下:

LooperTest: Not on main thread.

为什么会这样？我怀疑有一些奇怪的缓存机制在起作用。

此外，请注意 jobService.syncReservations() 返回一个 BehaviorSubject。通过各种搜索，似乎 Subjects 可能会或可能不会尊重对 .subscribeOn() 的调用。

最后，注意 jobServiceObservable 被注入(inject)到定义上述代码的文件中。 jobServiceObservable 是通过以下代码创建的:

public Observable<JobService> getObservable() {                                                                                              
    return Observable.create(e -> {                                                                                                          
        if (jobServiceBound && jobService != null) {                                                                                         
            e.onNext(jobService);                                                                                                            
            e.onComplete();                                                                                                                  
        } else {                                                                                                                             
            jobServiceConnection = new ServiceConnection() {                                                                                 
                @Override                                                                                                                    
                public void onServiceConnected(ComponentName name, IBinder service) {                                                        
                    JobService.JobServiceBinder binder = (JobService.JobServiceBinder) service;                                              
                    jobService = binder.getService();                                                                                        
                    jobServiceBound = true;                                                                                                  

                    e.onNext(jobService);                                                                                                    
                    e.onComplete();                                                                                                          
                }                                                                                                                            

                @Override                                                                                                                    
                public void onServiceDisconnected(ComponentName name) {                                                                      
                    reset();                                                                                                                 
                }                                                                                                                            
            };                                                                                                                               
            try {                                                                                                                            
                boolean success = context.bindService(new Intent(context, JobService.class), jobServiceConnection, Context.BIND_AUTO_CREATE);
                if (!success) {                                                                                                                                                                                    
                    e.onError(new Throwable("The service failed to be bound."));                                                             
                }                                                                                                                            
            } catch (SecurityException exception) {                                                                                                                                                                     
                e.onError(exception);                                                                                                        
            }                                                                                                                                
        }                                                                                                                                    
    });                                                                                                                                      
}

需要一个关于为什么会出现上述行为的权威答案。

Answer 1

因为在 io() 调度程序上订阅了包装 Observable 后，系统会在主线程上调用 onServiceConnected。 subscribeOn 告诉进一步订阅的位置，或者在这种情况下，Observable.create() 的主体应该执行。您应该在 flatMap 之前使用 observeOn 以便映射器函数在所需的线程上执行:

public Observable<Void> getObservable() {
    return jobServiceObservable
        .observeOn(Schedulers.io())
        .flatMap(jobService -> {
            if (Looper.myLooper() == Looper.getMainLooper()) {
                Log.d("LooperTest", "On main thread.");
            } else {
                Log.d("LooperTest", "Not on main thread.");
            }

            return jobService.syncReservations(
                accountUtil.getCurrentAccount()).subscribeOn(Schedulers.io());
        })
        .observeOn(foregroundScheduler);
}

(相比之下，对于典型的 Retrofit 网络调用，subscribeOn 之所以有效，是因为网络库在给定的调度程序上执行其阻塞调用并停留在那里以等待网络响应的发出。)