RxAndroid：Create Simple Hot Observable

Question

我創建了一個Observable，訂閱時發射整數。我的實現，現在被設置成訂閱它的行爲從一開始就觸發生成，如下所示：

private Observable createObservable() { 
    return Observable.create (
     new Observable.OnSubscribe<Integer>() { 

      @Override 
      public void call(Subscriber<? super Integer> sub) { 

       for (int i = 1; i < MAX_PROGRESS + 1; i++) { 
        sub.onNext(i); 
        SystemClock.sleep(1000); 
       } 
       sub.onCompleted(); 
      } 
     } 
    ); 
} 

我的理解是，這是一個寒冷的觀測。我希望序列能夠被生成而不管任何訂閱者，並且當訂閱者訂閱時，希望他們接收在訂閱時碰巧是最新的值。 IOW，把它變成一個熱門的Observable。我寧願不將Observable子類化，因爲它把它綁定到一個具體的Integer中，而實際上實際的類型會有所不同。

Answer 1

結賬rx.subjects.BehaviorSubject<T>。如果您對rx.subjects.Subject不熟悉，我認爲可以用它來描述它們的最普通的方式是，它們打破了A點和B點之間訂閱的連續性。它是怎麼一回事就是Observer<T>;可以接受多種來源的onNext()（警告：需要外螺紋安全）。另一方面，主題也是Observable<T>，所以多個Observer<T>可以訂閱，並且進入的onNext()將被多播到每個下游Observer<T>。

如果你的代碼看起來像

Observable<T> src = ...; 
Subscriber<T> dst; 
src.subscribe(dst); 

使用BehaviorSubject的方式是

Observable<T> src = ...; 
BehaviorSubject<T> subject = BehaviorSubject.create(defaultValue); 
src.subscribe(subject); 

立即訂閱源和主題將是儘可能快地被拋出。 BehaviorSubject只保留最近的值並刪除所有先前的默認值&。

// safe to do multiple times. 
Subscriber<T> dst; 
subject.subscribe(dst); 

在訂閱dst後立即訂閱接收從src（或defaultValue）中的最新值和直到dst取消訂閱然後所有後續值。

警告：受試者有過度使用的傾向，所以確保您需要一個。

Answer 2

在你的例子中，你每次調用函數時都使用「Observable.create」。 「熱」訂閱從Observable保存一些實例。你也需要用一些方法的Rx（緩存（），重試（））在代碼中它的樣子：

public Observable<Bitmap> mObservable; 
    Subscriber<Bitmap> mSubscriber; 
    Subscription mSubscription; 
    Bitmap loadedBitmap; 

.... 

@Override 
    public View onCreateView(LayoutInflater inflater, ViewGroup container, 
          Bundle savedInstanceState) { 

     if (savedInstanceState != null) { 
      loadedBitmap = savedInstanceState.getParcelable("LoadedBitmap"); 
      imageView.setImageBitmap(loadedBitmap); 
     } 
     else { 
      runNewObservable(); 
     } 
     runSubscribe(); 
     return mainView; 
    } 

.... 

    private void runNewObservable() { 
     mObservable = 
       Observable.create(new Observable.OnSubscribe<Bitmap>() { 
      @Override 
      public void call(Subscriber<? super Bitmap> subscriber) { 
       subscriber.onNext(new LoadingImage().loadImageFrom(imageURL)); 
       subscriber.onCompleted(); 
      } 
     }) 
       .subscribeOn(Schedulers.newThread()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .cache(); 

    } 



    private void runNewSubscribe() { 
     mSubscriber = new Subscriber<Bitmap>() { 

      @Override 
      public void onCompleted() { } 

      @Override 
      public void onError(Throwable e) { } 

      @Override 
      public void onNext(Bitmap bitmap) { 
       loadedBitmap = bitmap; 
       imageView.setImageBitmap(bitmap); 
      } 
     }; 
    } 

    @Override 
    public void onResume() { 
     super.onResume(); 
     mSubscription = mObservable.subscribe(mSubscriber); 
    } 


    @Override 
    public void onSaveInstanceState(Bundle outState) { 
     super.onSaveInstanceState(outState); 
     outState.putParcelable("LoadedBitmap", loadedBitmap); 
    } 



    @Override 
    public void onStop() { 
     super.onStop(); 
     mSubscription.unsubscribe(); 
    } 

.... 

正如你所看到的。我只使用mSubscription.subscribe（）和mSubscription.unsubscribe（），並僅在saveInstanceState爲null時創建Observable。