Java Streams — 如何每第 n 个项目执行一个中间函数

Question

我正在寻找一种对 Stream 的操作，它使我能够每隔 n 个项目执行一次非终端(和/或终端)操作。例如，虽然我使用了质数流，但流可以很容易地是 Web 请求、用户操作或其他一些冷数据或正在生成的实时提要。

来自这里:

    Duration start = Duration.ofNanos(System.nanoTime());

    IntStream.iterate(2, n -> n + 1)
            .filter(Findprimes::isPrime)
            .limit(1_000_1000 * 10)
            .forEach(System.out::println);

    System.out.println("Duration: " + Duration.ofNanos(System.nanoTime()).minus(start));

像这样的流函数:

    IntStream.iterate(2, n -> n + 1)
            .filter(Findprimes::isPrime)
            .limit(1_000_1000 * 10)
            .peekEvery(10, System.out::println)
            .forEach( it -> {});

Answer 1

创建一个辅助方法来包装 peek() 消费者:

public static IntConsumer every(int count, IntConsumer consumer) {
    if (count <= 0)
        throw new IllegalArgumentException("Count must be >1: Got " + count);
    return new IntConsumer() {
        private int i;
        @Override
        public void accept(int value) {
            if (++this.i == count) {
                consumer.accept(value);
                this.i = 0;
            }
        }
    };
}

您现在几乎可以完全按照自己的意愿使用它:

IntStream.rangeClosed(1, 20)
         .peek(every(5, System.out::println))
         .count();

输出

5
10
15
20

辅助方法可以放在实用程序类中并静态导入，类似于 Collectors 的方式。类不过是静态辅助方法。

@user140547 在 comment 中指出，此代码不是线程安全的，因此不能与并行流一起使用。此外，输出顺序会被打乱，所以无论如何将它与并行流一起使用是没有意义的。