RxJava:不调用Subscription的unsubscribe()方法

Question

在下面的代码中,我需要在unsubscription(它记录"release")上发布一些资源.

   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();

但它只记录"结果".看起来unsubscription没有传播到merge的子可观察量.那么如何处理第一个observable的Observable.OnSubscribe中的取消订阅？

Answer 1

大多数情况下,调用unsubscribe仅对实时序列有影响,并且如果某些序列已完成,则可能无法传播:运算符可能无法保留其源,因此可以避免内存泄漏.主要想法是运营商在他们调用下游onError或onCompleted方法之前或之后释放他们在终止时管理的任何资源,但这与1.x有些不一致.

如果您想确保资源是版本,请查看using将在终止或取消订阅时释放您的资源的运算符:

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