限速可观测

Question

我有一个可观察的对象，它从数据库游标的快速流中生成数据。我希望以每秒x项的速度限制输出。到目前为止，我一直在使用文档中所述的调用堆栈阻塞：

observable.map(f -> {
ratelimiter.acquire(); // configured limiter to only allow
});

这很好，但出于好奇，是否有更好的方法使用背压来处理此问题？

Tks公司

Answer 1

来自@michalsamek的答案似乎是正确的，尽管背压仅适用于Flowables。我已经更正了他的订阅者，以便它满足要求。

在不同的时间连续使用时，也有一个小问题。

private static <T> FlowableOperator<T, T> allowPerMillis(int millis) {
    return observer -> new PeriodicallyRequestingSubscriber<>(observer, millis);
}


Observable.range(1, 100)
    .observeOn(Schedulers.io())
    .toFlowable(BackpressureStrategy.BUFFER)
    .compose(Flowable::onBackpressureBuffer)
    .lift(allowPerMillis(200))
    .subscribe(value -> System.out.println(System.currentTimeMillis() % 10_000 + " : " + value));



public class PeriodicallyRequestingSubscriber<T> implements Subscriber<T> {

    private final Subscriber<T> upstream;

    private final int millis;

    // If there hasn't been a request for a long time, do not flood
    private final AtomicBoolean shouldRequest = new AtomicBoolean(true);

    public PeriodicallyRequestingSubscriber(Subscriber<T> upstream, int millis) {
        this.upstream = upstream;
        this.millis = millis;
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        Observable
                .interval(millis, TimeUnit.MILLISECONDS)
                .subscribe(x -> {
                   if (shouldRequest.getAndSet(false))
                       subscription.request(1);
                });
    }

    @Override
    public void onNext(T t) {
        shouldRequest.set(true);
        upstream.onNext(t);
    }

    @Override
    public void onError(Throwable throwable) {
        upstream.onError(throwable);
    }

    @Override
    public void onComplete() {
        upstream.onComplete();
    }
}

Answer 2

您可以尝试使用rx。可观察的#onBackpressureBuffer（）与自定义订阅者组合，该订阅者将每秒定期请求n个项目。但是，您必须进行一秒钟的采样。

注意。subscribeOn（）和。toBlocking（）只是为了使main方法不会立即退出。

public class BackpressureTest {

  public static void main(final String[] args) {
    Observable.range(1, 1000)
      .compose(Observable::onBackpressureBuffer) // consume source immediately, but buffer it
      .lift(allowPerSecond(3)) // via operator using custom subscriber request n items per second
      .subscribeOn(Schedulers.computation())
      .toBlocking()
      .subscribe(System.out::println);
  }

  private static <T> Observable.Operator<T, T> allowPerSecond(final int n) {
    return upstream -> periodicallyRequestingSubscriber(upstream, n);
  }

  private static <T> Subscriber<T> periodicallyRequestingSubscriber(final Subscriber<T> upstream, final int n) {
    return new Subscriber<T>() {

      @Override
      public void onStart() {
        request(0); // request 0 so that source stops emitting
        Observable.interval(1, SECONDS).subscribe(x -> request(n)); // every second request n items
      }

      @Override
      public void onCompleted() {
        upstream.onCompleted();
      }

      @Override
      public void onError(final Throwable e) {
        upstream.onError(e);
      }

      @Override
      public void onNext(final T integer) {
        upstream.onNext(integer);
      }
    };
  }
}

Answer 3

使用example（slttleLast）运算符：

Observable<T> throttled = 
    observable.sample(1 / rate, TimeUnit.MILLISECONDS);

http://reactivex.io/documentation/operators/sample.html

https://github.com/ReactiveX/RxJava/wiki/Backpressure