如何取消訂閱實時流

Question

我想了解如何取消訂閱從活動訂閱源創建的Observable。 這裏或多或少代碼：

SomeFeed feed = new SomeFeed(); 
Observable<PriceTick> observable = Observable.create(s -> 
    feed.register(new SomeListener() { 
    @Override 
    public void priceTick(PriceTick event) { 
     s.onNext(event); 
    } 

    @Override 
    public void error(Throwable throwable) { 
     s.onError(throwable); 
    } 
    }) 
); 
Subscription subscription = observable.subscribe(System.out::println); 
subscription.unsubscribe(); 
System.out.println("Is unsubscribed:" + subscription.isUnsubscribed()); // prints true 

我發現後認購退訂是，用戶仍然輸出事件流。

如何取消訂閱以刪除訂閱者接收更多通知？

Answer 1

您正在創建的可觀察項不服從Observable合約。你沒有處理退訂和退款。 Observable.create可以很容易地創建這樣的observables。

相反，您可以使用其他技術來創建您的observables。您可以使用defer運營商或Observable.fromCallable創建觀測值。

爲了您的代碼，你可以使用：

Observable<PriceTick> observable = Observable.create(s -> 
    feed.register(new SomeListener() { 
    @Override 
    public void priceTick(PriceTick event) { 
     if(!s.isUnsbscribed()) { 
     s.onNext(event); 
     } 
    } 

    @Override 
    public void error(Throwable throwable) { 
     if(!s.isUnsbscribed()) { 
     s.onError(throwable); 
     } 
    } 
    }) 
); 

這仍然不處理backpreasure。

更多關於創建觀測值here。