RxJava：在每个订阅者处理资源后关闭它

Question

我是RxJava的新手，我正在努力弄清楚如何正确关闭资源，尤其是在处理多个订阅者时。

我有一个<代码>可观察

我可能在observable上有多个订户。我想在每个订阅者处理完资源后，close（）resource。换句话说，在新资源被交付/发出后关闭旧资源，并且在最后一个订户取消订阅时也最终关闭最后一个资源。

我尝试使用一个自定义操作符让它工作，我称之为< code>AutoCloseOperator，它几乎可以工作了，但是不太正确。也就是说，我仍然处于竞争状态和/或泄漏，例如，资源没有关闭。

在RxJava中执行此操作的正确方法是什么？

假设我得到了这个代码:

final AutoCloseOperator<MyResource> autoClose = new AutoCloseOperator<MyResource>();
Subject<MyResource, MyResource> subject = PublishSubject.create();
Observable<MyResource> o = subject.lift(autoClose);

Subscription s1 = o.subscribe(new Action1<MyResource>() {
    public void call(MyResource myObj) {
        System.out.println("s1 handling " + myObj);
    }
});

subject.onNext(new MyResource(1));
subject.onNext(new MyResource(2)); // This should close Resource #1 after Resource #2 is delivered

Subscription s2 = o.subscribe(new Action1<MyResource>() {
    public void call(MyResource myObj) {
        System.out.println("s2 handling " + myObj);
    }
});

subject.onNext(new MyResource(3));
subject.onNext(new MyResource(4));

s1.unsubscribe();

subject.onNext(new MyResource(5));
subject.onNext(new MyResource(6));

s2.unsubscribe();

subject.onNext(new MyResource(7));
subject.onNext(new MyResource(8));

然后，我希望出现以下行为：

s1 handling Resource #1
s1 handling Resource #2
Closing Resource #1
s1 handling Resource #3
Closing Resource #2
s2 handling Resource #3
s1 handling Resource #4
s2 handling Resource #4
Closing Resource #3
s2 handling Resource #5
Closing Resource #4
s2 handling Resource #6
Closing Resource #5
Closing Resource #6
Closing Resource #7
Closing Resource #8

注意：在我的真实代码中，我没有使用＜code＞PublishSubject，每次更新数据库表时都会发出一个＜code＞光标。。。

为了概括问题：我可以使用 doOnNext 和 doOnUnsubition 来关闭旧项目，但这并没有考虑到这些事件将发生多次（对于每个订阅者），我只想在所有订阅者都收到新项目时关闭资源。

使用＜code＞lift（）

我已经将我的问题简化为GitHub上的一个小型命令行应用程序。感谢您的关注！

Answer 1

是可观察的。使用（）是您所需要的。

如果你有T类型的t，它有一个.close（）方法，并且你想从t（你的光标）中提取一些东西，说可观察

Func0<T> resourceFactory = () -> t;
Func1<T, Observable<R>> observableFactory = x -> ...
Action1<T> disposeAction = x -> x.close();

Observable<R> results = Observable.using(resourceFactory, observableFactory, disposeAction);

您提到您有可观察性

Observable<T> source = ...
Observable<R> results = 
    source.flatMap(t -> {
        Func0<T> resourceFactory = () -> t;
        Func1<T, Observable<R>> observableFactory = x -> ...
        Action1<T> disposeAction = x -> x.close();
        return Observable.using(resourceFactory, observableFactory, disposeAction);});