本文整理了Java中io.reactivex.Flowable.throttleFirst()方法的一些代码示例，展示了Flowable.throttleFirst()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Flowable.throttleFirst()方法的具体详情如下：
包路径：io.reactivex.Flowable
类名称：Flowable
方法名：throttleFirst

Flowable.throttleFirst介绍

[英]Returns a Flowable that emits only the first item emitted by the source Publisher during sequential time windows of a specified duration.

This differs from #throttleLast in that this only tracks the passage of time whereas #throttleLast ticks at scheduled intervals.

Backpressure: This operator does not support backpressure as it uses time to control data flow. Scheduler: throttleFirst operates by default on the computation Scheduler.
[中]返回在指定持续时间的连续时间窗口内仅发出源发布服务器发出的第一项的可流动项。
这与#throttleLast的不同之处在于，它只跟踪时间的流逝，而#throttleLast以预定的间隔滴答作响。
背压：此运算符不支持背压，因为它使用时间来控制数据流。调度程序：默认情况下，throttleFirst在计算调度程序上运行。

代码示例

代码示例来源：origin: ReactiveX/RxJava

@Test(expected = NullPointerException.class)
public void throttleFirstSchedulerNull() {
  just1.throttleFirst(1, TimeUnit.SECONDS, null);
}

代码示例来源：origin: ReactiveX/RxJava

@Test(expected = NullPointerException.class)
public void throttleFirstUnitNull() {
  just1.throttleFirst(1, null, Schedulers.single());
}

代码示例来源：origin: ReactiveX/RxJava

/**
 * Returns a Flowable that emits only the first item emitted by the source Publisher during sequential
 * time windows of a specified duration.
 * <p>
 * This differs from {@link #throttleLast} in that this only tracks the passage of time whereas
 * {@link #throttleLast} ticks at scheduled intervals.
 * <p>
 * <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/throttleFirst.png" alt="">
 * <dl>
 *  <dt><b>Backpressure:</b></dt>
 *  <dd>This operator does not support backpressure as it uses time to control data flow.</dd>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code throttleFirst} operates by default on the {@code computation} {@link Scheduler}.</dd>
 * </dl>
 *
 * @param windowDuration
 *            time to wait before emitting another item after emitting the last item
 * @param unit
 *            the unit of time of {@code windowDuration}
 * @return a Flowable that performs the throttle operation
 * @see <a href="http://reactivex.io/documentation/operators/sample.html">ReactiveX operators documentation: Sample</a>
 * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Backpressure">RxJava wiki: Backpressure</a>
 */
@CheckReturnValue
@BackpressureSupport(BackpressureKind.ERROR)
@SchedulerSupport(SchedulerSupport.COMPUTATION)
public final Flowable<T> throttleFirst(long windowDuration, TimeUnit unit) {
  return throttleFirst(windowDuration, unit, Schedulers.computation());
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void dispose() {
  TestHelper.checkDisposed(Flowable.just(1).throttleFirst(1, TimeUnit.DAYS));
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void testThrottlingWithCompleted() {
  Flowable<String> source = Flowable.unsafeCreate(new Publisher<String>() {
    @Override
    public void subscribe(Subscriber<? super String> subscriber) {
      subscriber.onSubscribe(new BooleanSubscription());
      publishNext(subscriber, 100, "one");    // publish as it's first
      publishNext(subscriber, 300, "two");    // skip as it's last within the first 400
      publishNext(subscriber, 900, "three");   // publish
      publishNext(subscriber, 905, "four");   // skip
      publishCompleted(subscriber, 1000);     // Should be published as soon as the timeout expires.
    }
  });
  Flowable<String> sampled = source.throttleFirst(400, TimeUnit.MILLISECONDS, scheduler);
  sampled.subscribe(subscriber);
  InOrder inOrder = inOrder(subscriber);
  scheduler.advanceTimeTo(1000, TimeUnit.MILLISECONDS);
  inOrder.verify(subscriber, times(1)).onNext("one");
  inOrder.verify(subscriber, times(0)).onNext("two");
  inOrder.verify(subscriber, times(1)).onNext("three");
  inOrder.verify(subscriber, times(0)).onNext("four");
  inOrder.verify(subscriber, times(1)).onComplete();
  inOrder.verifyNoMoreInteractions();
}

代码示例来源：origin: redisson/redisson

/**
 * Returns a Flowable that emits only the first item emitted by the source Publisher during sequential
 * time windows of a specified duration.
 * <p>
 * This differs from {@link #throttleLast} in that this only tracks the passage of time whereas
 * {@link #throttleLast} ticks at scheduled intervals.
 * <p>
 * <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/throttleFirst.png" alt="">
 * <dl>
 *  <dt><b>Backpressure:</b></dt>
 *  <dd>This operator does not support backpressure as it uses time to control data flow.</dd>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code throttleFirst} operates by default on the {@code computation} {@link Scheduler}.</dd>
 * </dl>
 *
 * @param windowDuration
 *            time to wait before emitting another item after emitting the last item
 * @param unit
 *            the unit of time of {@code windowDuration}
 * @return a Flowable that performs the throttle operation
 * @see <a href="http://reactivex.io/documentation/operators/sample.html">ReactiveX operators documentation: Sample</a>
 * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Backpressure">RxJava wiki: Backpressure</a>
 */
@CheckReturnValue
@BackpressureSupport(BackpressureKind.ERROR)
@SchedulerSupport(SchedulerSupport.COMPUTATION)
public final Flowable<T> throttleFirst(long windowDuration, TimeUnit unit) {
  return throttleFirst(windowDuration, unit, Schedulers.computation());
}

代码示例来源：origin: ReactiveX/RxJava

@Test
  public void backpressureNoRequest() {
    Flowable.range(1, 3)
    .throttleFirst(1, TimeUnit.MINUTES)
    .test(0L)
    .assertFailure(MissingBackpressureException.class);
  }
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void testThrottlingWithError() {
  Flowable<String> source = Flowable.unsafeCreate(new Publisher<String>() {
    @Override
    public void subscribe(Subscriber<? super String> subscriber) {
      subscriber.onSubscribe(new BooleanSubscription());
      Exception error = new TestException();
      publishNext(subscriber, 100, "one");    // Should be published since it is first
      publishNext(subscriber, 200, "two");    // Should be skipped since onError will arrive before the timeout expires
      publishError(subscriber, 300, error);   // Should be published as soon as the timeout expires.
    }
  });
  Flowable<String> sampled = source.throttleFirst(400, TimeUnit.MILLISECONDS, scheduler);
  sampled.subscribe(subscriber);
  InOrder inOrder = inOrder(subscriber);
  scheduler.advanceTimeTo(400, TimeUnit.MILLISECONDS);
  inOrder.verify(subscriber).onNext("one");
  inOrder.verify(subscriber).onError(any(TestException.class));
  inOrder.verifyNoMoreInteractions();
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void throttleFirstDefaultScheduler() {
  Flowable.just(1).throttleFirst(100, TimeUnit.MILLISECONDS)
  .test()
  .awaitDone(5, TimeUnit.SECONDS)
  .assertResult(1);
}