[reactor-06]flux- buffer - Publisher,Duration

计算机就跟比基尼一样，省去了人们许多的胡思乱想。

buffer(Publisher<?> other)

buffer(Publisher<?> other, Supplier<C> bufferSupplier)

根据other信号来决定每次缓存的个数, 当other.onNext的时候把缓存提交到消费者的onNext

public final Flux<List<T>> buffer(Publisher<?> other)
public final <C extends Collection<? super T>> Flux<C> buffer(Publisher<?> other, Supplier<C> bufferSupplier)

  public void test5() throws InterruptedException {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6).delayElements(Duration.ofSeconds(2));
        Flux<List<Integer>> buffer = flux.buffer(Duration.ofSeconds(3));
        buffer.subscribe(integers -> {
            System.out.println("产品到达,进行卸货:");
            integers.stream().forEach(integer -> {
                System.out.println(integer);
            });
        });

        Thread.sleep(20000);
    }

buffer(Duration bufferingTimespan)

buffer(Duration bufferingTimespan, Scheduler timer)

这两个方法是上面两个方法的应用, 传入的other是FluxInterval , 周期性触发onNext

示意图说明的很清楚.