FlowableOnBackpressureDrop忽略订阅的请求方法

Question

我在下面试过了

Flowable.interval(100L, TimeUnit.MILLISECONDS)
    .onBackpressureDrop()
    .observeOn(Schedulers.computation())
    .subscribe(new Subscriber<Long>() {

      private Subscription subscription;

      @Override
      public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        this.subscription.request(1L);
      }

      @Override
      public void onNext(Long t) {
        try {
          Thread.sleep(300L);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }

        System.out.println(t);

        subscription.request(1L);
      }

      ...
    });

我以为我会得到0，3，6...但是，我得到了0,1,2,3...因为Flowable最初得到的是请求(long.max_value)。

    @Override
    public void onSubscribe(Subscription s) {
        if (SubscriptionHelper.validate(this.s, s)) {
            this.s = s;
            actual.onSubscribe(this);
            s.request(Long.MAX_VALUE);
        }
    }

Answer 1

observeon请求并预先缓冲128个元素，因此在开始的一段时间内不会丢弃任何东西。如果您让它运行128*300ms，它将开始跳过值。您可以将带有重载的预取量设置为1，并看到值迅速下降。