執行請求或訂閱結果與RxJava

Question

我試圖實現與RxJava以下邏輯：

1. 執行服務器發出請求，如果我沒有價值本地
2. 結果提供給用戶
3. 如果請求正在運行，請不要創建第二個請求，但請訂閱運行請求的結果。

下面的解決方案部分地解決了這個問題：

private final ExecutorService executor = Executors.newSingleThreadExecutor(); 

private Observable<T> getValue(){ 
    if(storage.getValue() == null) { 
     Future<T> futureValue = executor.submit(new Callable<T>() { 
      @Override 
      public T call() throws Exception { 
       return getValueFromStorageOrBackend(); 
      } 
     }); 
     return Observable.from(futureValue); 
    } else { 
     return Observable.just(storage.getValue()); 
    } 
} 

private String getValueFromStorageOrBackend() { 
    final String currentValue = storage.getValue(); 
    if (currentValue == null) { 
     Response response = backend.requestValue(); 
     storage.setValue(response.getValue()); 

     return response.getValue(); 
    } 

    return currentValue; 
} 

對此有一個優雅的純RxJava解決方案？

Answer 1

其中一個簡單的方法來做到這一點是通過AsyncSubject +一個比較並設置循環：

final AtomicReference<AsyncSubject<T>> subjectRef = new AtomicReference<>(); 

public Observable<T> get() { 
    for (;;) { 
     AsyncSubject<T> subject = subjectRef.get(); 
     if (subject != null) { 
      return subject; 
     } 
     AsyncSubject<T> next = AsyncSubject.create(); 
     if (subjectRef.compareAndSet(null, next)) { 
      Observable.just(1).map(ignored -> { 
       // your computation here 
       return 2; // the result 
      }).subscribeOn(Schedulers.io()).subscribe(subject); 
      return; 
     } 
    } 
} 

我差點忘了，一個更簡單的方法是使用cache：

Observable 
.just(1) 
.subscribeOn(Schedulers.io()) 
.map(ignored -> { 
    // computation here 
    return 2; 
}).cache();