RxJava订阅阻挡观察到

Question

我想实现以下目标：

String result = myObservable.toBlocking().first(); 

即，它像一个普通的函数调用。然而，这从来不会发生，因为你需要订阅它，我不知道该怎么做。如果我订阅它，结果将在另一个范围内，并且代码非常难看，因为我只能得到像常规观察值一样的结果，所以没有必要将它变成阻塞的可观察值。

Answer 1

它的实际工作，只要你想：

Observable<String> myObservable = Observable.just("firstValue", "secondValue"); 
    String result = myObservable.toBlocking().first(); 
    System.out.println(result); // ---> "firstValue" 

引擎盖下，调用BlockingObservable.first()并订阅为您提供：

private T blockForSingle(final Observable<? extends T> observable) { 
    final AtomicReference<T> returnItem = new AtomicReference<T>(); 
    final AtomicReference<Throwable> returnException = new AtomicReference<Throwable>(); 
    final CountDownLatch latch = new CountDownLatch(1); 

    @SuppressWarnings("unchecked") 
    Subscription subscription = ((Observable<T>)observable).subscribe(new Subscriber<T>() { 
     @Override 
     public void onCompleted() { 
      latch.countDown(); 
     } 

     @Override 
     public void onError(final Throwable e) { 
      returnException.set(e); 
      latch.countDown(); 
     } 

     @Override 
     public void onNext(final T item) { 
      returnItem.set(item); 
     } 
    }); 
    BlockingUtils.awaitForComplete(latch, subscription); 

    if (returnException.get() != null) { 
     Exceptions.propagate(returnException.get()); 
    } 

    return returnItem.get(); 
} 

UPDATE：如果没有任何意义使用一个BehaviourSubject加toBlocking()。考虑到它是和Observable和Observer所以在某处应该调用myObservable.onNext("value")。如果您通过调用toBlocking()来阻止该线程，除非myObservable在其他调用onNext()的其他线程中可用，否则您将被阻止。

举例来说，这是正常使用`BehaviourSubject的：

// observer will receive the "one", "two" and "three" events, but not "zero" 
    BehaviorSubject<Object> subject = BehaviorSubject.create("default"); 
    subject.onNext("zero"); 
    subject.onNext("one"); 
    subject.subscribe(observer); 
    subject.onNext("two"); 
    subject.onNext("three");