使用CONCAT（）と最初の（）RxJavaでキャッシュを実装する - 同時呼び出し

Question

私はRxJavaとconcat()とfirst()演算子を使用しています：

public Observable<List<Entity>> getEntities() { 
    invalidateCacheIfNeeded(); 
    return Observable 
      .concat(cachedEntities(), networkEntities()) 
      .first(); 
} 

cachedEntitiesをキャッシュされたリストの中から構築された観察可能を返しますnetworkEntitiesメソッドはRetrofitでエンティティを取得します。

これは、2人のユーザーがgetEntities()によって返されたオブザーバブルにすばやく登録しない限りうまくいきます。私は、最初の購読のネットワーク要求は、2番目の購読が行われたときに終了していないと思います。この場合、2つのネットワーク要求が実行されます。私は避けたいものです。

最初の呼び出しが終わったが、運にあるときに、第2のコールの実行のみが行われるので、私はシングルスレッドスケジューラを作成しようとしました：

mSingleThreadScheduler = Schedulers.from(Executors.newSingleThreadExecutor()); 

と：

public Observable<List<Entity>> getEntities() { 
    invalidateCacheIfNeeded(); 
    return Observable 
      .concat(cachedEntities(), networkEntities()) 
      .subscribeOn(mSingleThreadScheduler) 
      .first(); 
} 

ObservableチェーンでsubscribeOnコールを振り下ろそうとしましたが、同じ結果が得られました。

ヒント？

Answer 1

private Observable<List<Entity>> networkEntities() { 
    return mSubject 
      .map(Data::getEntities); 
} 

に従ってくださいそして、あなたのネットワーク呼び出しは、私はあなたが簡単に使用できると考えている。この

public Observable<Data> getDataFromNetwork() { 
    return networkOperation() 
      .subscribeOn(mSingleThreadScheduler) 
      .subscribe(mSubject); 
} 

Answer 2

スレッドセーフなメソッドを作るのは良い考えではないと思います。メソッド全体をブロックするため、パフォーマンスが低下します。そのため、データ構造をスレッドセーフにすることをお勧めします。あなたのケースで あなたはあなたの方法

public Observable<List<Entity>> getEntities() { 

}

使用CopyOnWriteArrayListと代わりの一覧に一覧を使用しています。スレッドセーフです。

public Observable<CopyOnWriteArrayList<Entity>> getEntities() { 

}

それが動作します願っています。

Answer 3

これは、複数の加入者が観察できる比較的一般的な使用例のようです。あなたは、深さの説明では、より ためthis questionへの回答を参照してください

public Observable<List<Entity>> getEntities() { 
    invalidateCacheIfNeeded(); 
    return Observable 
      .concat(cachedEntities(), networkEntities()) 
      .first() 
      .replay(1)    
} 

ようなものが必要。