無限並列ストリームの制限

Question

1）サプライヤ（supplier）を使用してN個の値のサイズ付きストリームを並行して作成し、サプライヤにN回以上のコールを行わないようにするにはどうすればよいですか？これは、私が高価なsupplier.get()操作のサプライヤを持っているために必要です。

2）私の質問Streams.generate(supplier).limit(N)の「明白な」答えはうまくいかず、しばしばサプライヤーにN回以上の呼び出しが行われます。どうしてこれなの？

public class MWE { 
    static final int N_ELEMENTS=100000; 
    static Supplier<IntSupplier> mySupplier =() -> new IntSupplier() { 
     AtomicInteger ai = new AtomicInteger(-1); 
     @Override 
     public int getAsInt() { 
      return ai.incrementAndGet(); 
     } 
    }; 
    public static void main(String[] args) { 
     int[] a = IntStream.generate(mySupplier.get()).limit(N_ELEMENTS).toArray(); 
     int[] b = IntStream.generate(mySupplier.get()).parallel().limit(N_ELEMENTS).toArray(); 
    } 
} 

a予想通り[0, 1, ..., N_ELEMENTS-1]に等しいが、あなたはbを期待するかもしれないものに反している：Nは、次のコードを検討し、supplier.get()への呼び出しよりも多くの中Streams.generate(supplier).limit(N)結果があるという事実の「証拠」として

aと同じ要素は含まれていません。代わりに、bには、N_ELEMENTS以上の要素が含まれている場合があり、これはN_ELEMENTS件以上のサプライヤへの呼び出し回数を示します。

Streams.generate(new Random(0)::nextDouble()).limit(5)が必ずしも同じ番号のセットを生成するとは限りません。

Answer 1

はStream.generate()を維持するために、ストリームの何の部分 "を決定する自由limit()を残しunorderedストリームを、作成されるため、供給者によって生成された第1のN要素のストリームをもたらすことが保証されていません。実際には、ストリームが順序付けられていないため、「ストリームの最初のN要素」または「（最初の）部分」を意味するのは意味的ではありません。この動作は、APIのドキュメントに明確に記載されています。これを私に指摘してくれた皆様に感謝します！

この質問をして以来、私は自分の質問に2つの解決策を思いついた。私は正しい方向に私を止めてくれたTagirに感謝します。

解決策1：IntStream.range()

絶対に必要であるよりもサプライヤーにこれ以上の呼び出しを行いません業者に支え順不同、サイズ、並列ストリームを作成する単純な、かなり効率的な方法を誤用を使用（MIS）でありますこのようなIntStream.range()：

IntStream.range(0,N_ELEMENTS).parallel().mapToObj($ -> generator.get()) 

基本的に、私たちは、並行して処理することが可能なサイズのストリームを作成するためにIntStream.range()を使用しています。

解決方法2：カスタムspliterator

我々は実際にIntStream.range()によって作成されたストリームの内部整数を使用することはありませんので、我々はカスタムSpliteratorを作成することによって、わずかに良い行うことができますように、それはそうです：

final class SizedSuppliedSpliterator<T> implements Spliterator<T> { 
    private int remaining; 

    private final Supplier<T> supplier; 

    private SizedSuppliedSpliterator(Supplier<T> supplier, int remaining) { 
     this.remaining = remaining; 
     this.supplier = supplier; 
    } 

    static <T> SizedSuppliedSpliterator of(Supplier<T> supplier, int limit) { 
     return new SizedSuppliedSpliterator(supplier, limit); 
    } 

    @Override 
    public boolean tryAdvance(final Consumer<? super T> consumer) { 
     Objects.requireNonNull(consumer); 
     if (remaining > 0) { 
      remaining--; 
      final T supplied = supplier.get(); 
      consumer.accept(supplied); 
      return true; 
     } 
     return false; 
    } 

    @Override 
    public void forEachRemaining(final Consumer<? super T> consumer) { 
     while (remaining > 0) { 
      consumer.accept(supplier.get()); 
      remaining--; 
     } 
    } 

    @Override 
    public SizedSuppliedSpliterator<T> trySplit() { 
     int split = (int)remaining/2; 
     remaining -= split; 
     return new SizedSuppliedSpliterator<>(supplier, split); 
    } 

    @Override 
    public long estimateSize() { 
     return remaining; 
    } 

    @Override 
    public int characteristics() { 
     return SIZED | SUBSIZED | IMMUTABLE; 
    } 
} 

このスプライテータを使用して、次のようにストリームを作成することができます。

もちろん、2つの整数を計算するのはコストがかかりません。解決方法1より性能が改善されていることに気づくことはできませんでした。

Answer 2

IntStream.generate()が指定された回数だけコールすることをストリームAPIが保証しません。また、この呼び出しは順序付けを考慮しません。

数値が増加する並列ストリームが実際に必要な場合は、IntStream.range(0, N_ELEMENTS).parallel()を使用する方がはるかに優れています。これにより、実際に0からN_ELEMENTS-1までのすべての数値が保証されるだけでなく、競合が大幅に軽減され、注文が保証されます。より複雑なものを生成する必要がある場合は、独自のSpliteratorクラスを定義するカスタムソースの使用を検討してください。

提案されているIntStream.iterateソリューションは、シーケンシャルバイソースのソースとして非常に並列化できないことに注意してください。呼び出し.limit()