我是 RxJava 的新手，我需要創(chuàng)建包含多個數(shù)據(jù)源的存儲庫。這對我來說很復雜，因為有幾個較小的子任務我不知道如何用 RxJava 實現(xiàn)。我有 Dao，它Flowable<Item>在一定范圍內(nèi)提供給 DataSource 類。該數(shù)據(jù)源有本地緩存，可以隨時失效。當存儲庫向 DataSource 詢問某個范圍時（可能超出 DataSourse 范圍，邊界在完全緩存之前是未知的）它必須產(chǎn)生錯誤（或以其他方式通知 Repository）。我想Flowable<Item>為 DataSource 創(chuàng)建方法，它將從緩存中發(fā)出項目，如果需要，將它們與連接起來Flowable<Item> dao.getRange(...)，同時緩存來自 dao 的新項目。我還需要處理來自 dao 的錯誤，它們必須被處理或轉換為更高級別的錯誤。DataSource.class:List<Item> cache;Flowable<Item> getRange(int start, int amount) {    final int cacheSize = cache.size();    final int canLoadFromCache = cacheSize - start;    final int loadFromDao = amount - canLoadFromCache;    if (isCorrupted) return Flowable.fromCallable(() -> {        throw new Exception("CorruptedDatasource");    });    Flowable<Item> cacheFlow = null;    Flowable<Item> daoFlow = null;    if (canLoadFromCache > 0) {        cacheFlow = Flowable.fromIterable(                cache.subList(start, canLoadFromCache)        );        daoFlow = dao.getRange(                uri,                 cacheSize, //start                loadFromDao //amount        );    } else {        if (isFullyCached) return Flowable.fromCallable(() -> {            throw new Exception("OutOfBounds");        });        //To not deal with gaps load and cache data between;        //Or replace it with data structure which can handle for us;        daoFlow = dao.getRange(                uri,                cacheSize,                start - cacheSize + amount);        //all these items should be cached;        //other cached and put downstream;        //Dao errs should be converted to higher lever exceptions,        //Or set flags in DataSource;    }    // return concatenated flowable;}在更高級別的存儲庫連接來自多個數(shù)據(jù)源的數(shù)據(jù)，因此必須有一種方法來連接來自多個來源的范圍，如果一個來源還不夠，則應該添加下一個來源的范圍。