如何将Completable与Observable结合？改造现有数据流加载逻辑

要在不改动原有天气数据流逻辑的前提下，实现订阅Observable时自动执行缓存API（Completable），可以根据你的实际需求选择以下两种方案：

方案一：订阅时异步执行缓存操作（不阻塞数据流） #

如果不需要等待缓存操作完成就发射天气数据，推荐使用doOnSubscribe操作符——它会在Observable被订阅时触发一次缓存API调用，完全不影响原有数据流的发射时机和内容：

fun fetchWeather(): Observable<Weather> {
    return weatherObservable
        .startWithItem(/* Weather Object */)
        .doOnSubscribe {
            // 订阅时执行缓存操作，可添加错误回调处理异常
            hasWeatherFetched().subscribe(
                {}, 
                { error -> 
                    // 处理缓存操作的异常，比如打印日志
                    Log.e("WeatherCache", "缓存操作失败", error) 
                }
            )
        }
}

如果缓存操作需要在特定线程执行（比如IO线程），可以给Completable指定线程调度：

fun fetchWeather(): Observable<Weather> {
    return weatherObservable
        .startWithItem(/* Weather Object */)
        .doOnSubscribe {
            hasWeatherFetched()
                .subscribeOn(Schedulers.io()) // 在IO线程执行缓存操作
                .observeOn(AndroidSchedulers.mainThread()) // 可选：回调到主线程处理结果
                .subscribe({}, { error -> Log.e("WeatherCache", "缓存失败", error) })
        }
}

方案二：等待缓存完成后再发射天气数据 #

如果要求必须先完成缓存操作，再发射天气数据，可以用andThen将Completable和原有Observable串联：

fun fetchWeather(): Observable<Weather> {
    // 先执行缓存操作，完成后再发射原有天气数据流
    return hasWeatherFetched()
        .andThen(weatherObservable.startWithItem(/* Weather Object */))
}

注意：这个方案会改变原有逻辑——原来的startWithItem会立即发射数据，现在要等缓存操作完成后才会发射，适合必须依赖缓存完成的场景。

内容的提问来源于stack exchange，提问作者Hack123