如何在ReactiveX中过滤过于频繁的onSubscribe请求

Question

我有一个执行耗时网络操作的可观察对象。客户端代码可能会频繁订阅可观察对象，从而导致高网络负载。

由于我们无法控制订阅者出现的时间，所以这必须在可观察对象端完成。

Answer 1

当并发订阅者数量达到最大值时，您希望其他订阅者接收空流。

给定一个想要限制订阅的源，这样做:

Observable<T> limited = source.compose(
        new TransformerLimitSubscribers<T>(
            new AtomicInteger(), maxSubscribers))
     .onErrorResumeNext(Observable.<T>empty());
...
limited.subscribe(s1);
...
limited.subscribe(s2); 

，其中转换器由以下类定义:

public final class TransformerLimitSubscribers<T> implements Transformer<T, T> {
    private final AtomicInteger subscriberCount;
    private final int maxSubscribers;
    public TransformerLimitSubscribers(AtomicInteger subscriberCount, int maxSubscribers) {
        this.subscriberCount = subscriberCount;
        this.maxSubscribers = maxSubscribers;
    }
    @Override
    public Observable<T> call(Observable<T> o) {
        return o.doOnSubscribe(onSubscribe()).doOnUnsubscribe(onUnsubscribe());
    }
    private Action0 onSubscribe() {
        return new Action0() {
            @Override
            public void call() {
                if (subscriberCount.incrementAndGet() > maxSubscribers)
                    throw new TooManySubscribersException();
            }
        };
    }
    private Action0 onUnsubscribe() {
        return new Action0() {
            @Override
            public void call() {
                subscriberCount.decrementAndGet();
            }
        };
    }
    public static class TooManySubscribersException extends RuntimeException {
    }
}