我有 2 個數據源：數據庫（緩存）和 api，我需要將它們組合成一個流。我知道我可以簡單地使用 concatArray 或類似的東西，但我想實現更復雜的行為：最多可發射 2 個元素的可觀察流。它將在開始時訂閱這兩個來源。如果 api 調用足夠快（<~300 毫秒），它將僅從中發出數據并完成流。如果 api 調用會很慢（>~300ms），從數據庫發出數據并仍然等待來自 api 的數據如果 api 調用不會成功，則從數據庫發出數據并發出錯誤。如果數據庫以某種方式比 api 慢，它就不能發出它的數據（流完成解決了這個問題）我用以下代碼完成了它：   public Observable<Entity> getEntity() {    final CompositeDisposable disposables = new CompositeDisposable();    return Observable.<Entity>create(emitter -> {        final Entity[] localEntity = new Entity[1];        //database call:        disposables.add(database.getEntity()                .subscribeOn(schedulers.io())                .doOnSuccess(entity -> localEntity[0] = entity) //saving our entity because                                                         //apiService can emit error before 300 ms                 .delay(300, MILLISECONDS)                .subscribe((entity, throwable) -> {                    if (entity != null && !emitter.isDisposed()) {                        emitter.onNext(entity);                    }                }));        //network call:        disposables.add(apiService.getEntity()                .subscribeOn(schedulers.io())                .onErrorResumeNext(throwable -> {                    return Single.<Entity>error(throwable) //we will delay error here                            .doOnError(throwable1 -> {                                if (localEntity[0] != null) emitter.onNext(localEntity[0]); //api error, emit localEntity                            })                            .delay(200, MILLISECONDS, true); //to let it emit localEntity before emitting error                })                .subscribe(entity -> {                    emitter.onNext(entity);                     emitter.onComplete(); //we got entity from api, so we can complete the stream                }, emitter::onError));    })代碼有點笨重，我在 observable 中創建了 observables，我認為這是不好的。但是這樣我就可以全局訪問發射器，這使我能夠以我想要的方式控制主流（發射數據、成功、錯誤）。有沒有更好的方法來實現這一目標？我很想看一些代碼示例。謝謝！