rx-java - RxJava2：第一个订阅和最后一个取消订阅的观察者的 PublishProcessor 回调

Question

我使用 aPublishProcessor<?>来服务多个观察者。有没有办法知道第一个观察者何时被订阅，最后一个观察者何时被处置？

Answer 1

不是直接的，你需要捕获订阅者和取消者：

PublishProcessor<?> pp = ...

AtomicInteger counter = new AtomicInteger();

Action onFirst = ...

Action onLast = ...

Flowable<?> f = pp.doOnSubscribe(s -> {
    if (counter.getAndIncrement() == 0) {
        onFirst.run();
    }
})
.doFinally(() -> {
    if (counter.decrementAndGet() == 0) {
        onLast.run();
    }
})

// use f for subscribe() instead of pp