反应堆3“间隔缓冲”流量？

Question

如何使用现有的Flux操作符使Flux将传入的值返回到多个列表中，并且返回之间的延迟最小？

​

​

Answer 1

这可以通过一组非平凡的复合运算符来实现。

import java.time.Duration;
import java.util.*;

import reactor.core.publisher.*;

public class DelayedBuffer {

    public static void main(String[] args) {
        Flux.just(1, 2, 3, 6, 7, 10)
        .flatMap(v -> Mono.delayMillis(v * 1000)
              .doOnNext(w -> System.out.println("T=" + v))
              .map(w -> v)
        )
        .compose(f -> delayedBufferAfterFirst(f, Duration.ofSeconds(2)))
        .doOnNext(System.out::println)
        .blockLast();
    }

    public static <T> Flux<List<T>> delayedBufferAfterFirst(Flux<T> source, Duration d) {
        return source
        .publish(f -> {
            return f.take(1).collectList()
            .concatWith(f.buffer(d).take(1))
            .repeatWhen(r -> r.takeUntilOther(f.ignoreElements()));
        });
    }
}

(但是，请注意，由于涉及时间，预期的发射模式可能与自定义操作符更好地匹配。)

Answer 2

我以为buffer(Duration)会满足您的需求，但它不是。

编辑:保留此选项，以防与您有完全相同需求的人尝试使用该操作符。buffer的这种变体将序列分成多个连续的时间窗口(每个窗口产生一个buffer)。也就是说，新的delay从前一个元素的末尾开始，而不是在发出新的超时元素时开始。