N 个元素中的样本 1

【问题标题】：Sample 1 in N elementsN 个元素中的样本 1
【发布时间】：2021-04-18 15:14:35
【问题描述】：

我有一个长期运行的 Flux，想在 N 个元素中记录 1 来监控进度。以下代码在 N 毫秒内记录一个。

Flux
    .fromStream(
        IntStream
            .range(1, 101)
            .mapToObj(Integer::valueOf)
    )
    .sample(Duration.ofMillis(2))
    .subscribe(e -> log.debug(e.toString()));

sample(Publisher...) 之类的声音可用于在 N 个元素中记录 1，方法是为第 1 个元素生成 Mono.Just("")，为其余元素生成 Mono.empty()。但是该方法不提供被采样的元素。请求有关如何解决此问题的想法？

【问题讨论】：

标签： project-reactor

【解决方案1】：

您可以使用Flux#index 获取索引 Flux 并记录每个第 n 个元素。这可以使用Flux#transform 干净地合并到您的链中。

log1InN实用方法：

<T> Flux<T> log1InN(Flux<T> source, int n) {
        return source.index()
                .doOnNext(e -> {
                            if (e.getT1() % n == 0) {
                                log.info("Element {}: {}", e.getT1(), e.getT2());
                            }
                        }
                ).map(Tuple2::getT2);
    }

然后用它来记录例如每十个元素：

Flux.fromStream(
                IntStream
                        .range(1, 101)
                        .mapToObj(Integer::valueOf)
        )
                .transform(f -> log1InN(f, 10))
                .subscribe();

【讨论】：