RxJava：未调用订阅的 unsubscribe() 方法答案

【问题标题】：RxJava: Subscription's unsubscribe() method isn't invokedRxJava：未调用订阅的 unsubscribe() 方法
【发布时间】：2015-11-02 09:21:30
【问题描述】：

在下面的代码中，我需要在取消订阅时释放一些资源（它记录“释放”）。

   Observable first = Observable.create(new Observable.OnSubscribe<Object>() {
                    @Override
                    public void call(Subscriber<? super Object> subscriber) {
                        subscriber.add(Subscriptions.create(() -> {
                            log(“release”);         
                        }));
                    }
                }).doOnUnsubscribe(() -> log(“first”));
   Observable second = Observable.create(…).doOnUnsubscribe(() -> log(“second”));
   Observable result = first.mergeWith(second).doOnUnsubscribe(() -> log(“result”));
   Subscription subscription = result.subscribe(…);
   //…
   subscription.unsubscribe();

但它只记录“结果”。看起来取消订阅不会传播到合并的子可观察对象。那么如何在第一个 observable 的 Observable.OnSubscribe 中处理取消订阅呢？

【问题讨论】：

标签： java reactive-programming rx-java rx-android

【解决方案1】：

大多数情况下，调用unsubscribe 仅对实时序列有影响，如果某些序列已完成，则可能不会传播：操作员可能不会保留其源代码，以避免内存泄漏。主要思想是运营商在调用下游的onError 或onCompleted 方法之前或之后释放他们在终止时管理的任何资源，但这与1.x 有点不一致。

如果您想确保资源被释放，请查看using 运算符，它将在终止或取消订阅时释放您的资源：

Observable.using(
    () -> "resource", 
    r -> Observable.just(r), 
    r -> System.out.println("Releasing " + r))
.subscribe(System.out::println);

【讨论】：