スプリッターを書く

皆さんの多くはすでにストリームAPI-Java 8ストリームを味わっていますが、コレクション、配列、乱数からの既製のストリームを使用するだけでなく、根本的に新しいストリームを作成したい人もいます。 これを行うには、スプリッターを作成する必要があります。 Spliteratorは、内部ロジックのパブリック部分であるストリームを埋めます。 この記事では、スプリッターを作成した方法と理由を説明します。

スプリッターとは

スプリッターは8つのメソッドを含むインターフェースであり、そのうちの4つはすでにデフォルトの実装を持っています。 残りのメソッドは、 tryAdvance 、 trySplit 、 estimateSizeおよびtrySplitです。 プリミティブ型int 、 longおよびdouble特別なスプリッター変更もあります。これらは、ボクシングを回避するためにいくつかの追加メソッドを追加します。 スプリッターは通常のイテレーターのようなものです。 主な違い-2つの部分に分割（分割）する機能-は、スレッドの並列操作の基本です。 また、最適化するために、スプリッターには多数のフラグ特性があり、正確にまたはほぼそのサイズを報告できます。 最後に、スプリッターは決してデータソースを変更しません。イテレーターのようなremoveメソッドはありません。 より詳細にメソッドを検討してください。

• tryAdvance（コンシューマー） hasNext()およびnext()イテレーターメソッドのhasNext() 。 スプリッターに次の要素がある場合、渡された関数をこの要素で呼び出してtrueを返す必要がありfalse 。そうでない場合、関数は呼び出されずにfalseを返しfalse 。
• trySplit（） -2つの共有を試みます。 このメソッドは、元のデータセットの前半を実行する新しいスプリッターを返しますが、現在のスプリッター自体は後半にジャンプします。 半分がほぼ等しい場合に最適ですが、これは必要ありません。 無限のデータセットを持つスプリッターは、特に不均等に分割されます。分割後、スプリッターの1つが最終ボリュームを処理し、2番目のスプリッターは無限に残ります。 trySplit()メソッドには、 nullを共有して返さないという法的権利がありnull （偶然にtryすることはありません）。 これは通常、現在のスプリッターにデータがほとんど残っていない場合（たとえば、1つの要素のみ）に行われます。
• 特性（） -スプリッターの特性のビットマスクを返します。 現在、そのうちの8つがあります。
  
  1. ORDERED-データの順序が重要な場合。 たとえば、 HashSet内のデータの順序は実装に依存するため、 HashSetスプリッターにはこの特性がありません。 この特性がないと、パラレルストリームが自動的に無秩序モードに移行するため、動作が速くなります。 データソースに順序がなかったため、それに従うことはできません。
  2. DISTINCT-要素が明らかに一意である場合。 distinct()操作の後のSetまたはストリームは、この特性を持つスプリッターを作成します。 たとえば、 Setからのストリームに対するdistinct()操作はまったく実行されないため、時間がかかりすぎません。
  3. SORTED-アイテムがソートされている場合。 この場合、ORDEREDの両方を返し、 getComparator()メソッドをオーバーライドして、ソートコンパレータまたは「 自然順序 」の場合はnullを返す必要があります。 ソートされたコレクション（ TreeSet ）は、この特性を持つスプリッターを作成します。これにより、 sorted()たストリーム操作sorted()をスキップできます。
  4. SIZED-スプリッター要素の正確な数がわかっている場合。 この特性は、すべてのコレクションのスプリッターによって返されます。 一部のストリーム操作（たとえば、 map()またはsorted() ）の後、保存され、他のストリーム操作（たとえば、 filter()またはdistinct() ）後に失われます。 これはソートや、たとえばtoArray()操作に役立ちます。適切なサイズの配列を事前に選択でき、必要な要素の数を推測できません。
  5. SUBSIZED-すべての子スプリッターもサイズを認識することがわかっている場合。 この特性はArrayListからスプリッターによって返されます。これは、分割するときに値の範囲を既知の長さの2つの範囲に単純に分割するためArrayList 。 しかし、 HashSetはそれを返しません。これは、ハッシュテーブルを壊すためです。このため、各半分にいくつの要素があるかはわかりません。 したがって、子スプリッターはSIZEDを返しません。
  6. NONNULL-要素間にnullがないことがわかっている場合。 この特性は、たとえば、 ConcurrentSkipListSetによって作成されたスプリッターによって返されnull 。このデータ構造にnullを配置することはできません。 また、プリミティブ型で作成されたすべてのスプリッターによって返されます。
  7. IMMUTABLE-クロール中にデータソースを変更できないことがわかっている場合。 通常のコレクションからのスプリッターはそのような特性を返しませんが、たとえば、 Collections.singletonList()からのスプリッターはこのリストを変更できないため、それを提供します。
  8. 並行-ソースへの変更後もスプリッタが動作し続けることがわかっている場合。 この特性は、 java.util.concurrentコレクションスプリッターによって報告されます。 スプリッタにIMMUTABLEおよびCONCURRENT特性がない場合、ソースが変更されたことに気付いた場合にConcurrentModificationExceptionスローするように、フェイルファーストモードで動作させると便利です。
  私の知る限り、最後の3つの特性はスレッド（Java 9コードを含む）で現在使用されていません。
• guessSize （） -メソッドは、SIZEDスプリッターの残りの要素数と、他の場合に可能な限り最も正確な推定値を返す必要があります。 たとえば、 HashSetからスプリッターを作成し、 trySplit()を使用してスプリッターを分割すると、 estimateSize()サイズestimateSize()はコレクションの元のサイズの半分を返しますが、ハッシュテーブルの半分の実際の要素数は異なる場合があります。 要素の数が無限にある場合、または要素のカウントがLong.MAX_VALUEすぎる場合は、 Long.MAX_VALUEを返すことができLong.MAX_VALUE 。

既存のスプリッターを使用してストリームを作成するのは非常に簡単です-StreamSupport.stream（）を呼び出す必要があります。

スプリッターを書く必要がないとき

理解すべき主なことは、スプリッター自体は必要なく、ストリームが必要であることです。 既存の機能を使用してストリームを作成できる場合は、それを行う必要があります。 たとえば、XML DOMストリームと友達になり、 NodeListを使用してストリームを作成するとします。 このような標準的な方法はありませんが、追加のスプリッタなしで簡単に記述できます。

 public class XmlStream { static Stream<Node> of(NodeList list) { return IntStream.range(0, list.getLength()).mapToObj(list::item); } } 

同様に、任意の非標準コレクション（別の例はorg.json.JSONArray ）にストリームを追加できます。これにより、序数で長さと要素をすばやく返すことができます。

trySplitを書くのが難しいか怠けているとtrySplit場合は、スプリッターをまったく書かない方が良いです。 次に、プロトンスレッドライブラリを記述し、並列スレッドの存在を完全に無視する仲間を示します 。 彼は、共有方法がまったくわからないスプリッターをたくさん書いた。 まったく分割しないスプリッターは、悪い、価値のないスプリッターです。 そうしないでください。 この場合、通常のイテレーターを記述し、 Spliterators.spliteratorメソッドを使用してその上にスプリッターを作成するか、事前にコレクションのサイズがわからない場合はSpliterators.spliteratorUnknownSizeを作成することをお勧めします。 これらのメソッドには、少なくとも除算のヒューリスティックがあります。反復子の一部をバイパスし、それを配列に減算して、この配列の新しいスプリッターを作成します。 ストリームが長時間の操作を継続する場合、並列化により作業が加速されます。

標準のIterableまたはCollectionインターフェイスを実装すると、 default spliterator() defaultメソッドが提供されます。 もちろん、改善できるかどうかを確認する価値はあります。 何らかの方法で、スプリッターを作成する必要はほとんどありません。 これは、データ構造（たとえば、 leventovのようにプリミティブのコレクション）を開発している場合に便利です。

そしてまだ書く

この問題を解決するために、新しいスプリッターを作成します。特定のストリームのソースストリームの隣接する値からペアのストリームを作成します。 Javaで値のペアを表すために一般に受け入れられているタイプはなく、可能なオプションが多すぎるため（2つの値の配列、2つの値のリスト、同じタイプのキーと値を持つMap.Entryなどを使用）、ユーザーに提供します：2つの値を組み合わせる方法を彼に決めさせます。 つまり、次のシグネチャを持つメソッドを作成する必要があります。

 public static <T, R> Stream<R> pairMap(Stream<T> stream, BiFunction<T, T, R> mapper) {...} 

これにより、任意のタイプを使用してペアを表すことができます。 たとえば、 Map.Entry必要な場合：

 public static <T> Stream<Map.Entry<T, T>> pairs(Stream<T> stream) { return pairMap(stream, AbstractMap.SimpleImmutableEntry<T, T>::new); } 

そして、一般的に、中間コンテナにペアを積み重ねることなく、すぐに興味深いものを計算できます。

 public static Stream<Integer> diff(Stream<Integer> stream) { return pairMap(stream, (a, b) -> b - a); } 

このメソッドは、整数のストリームによる隣接要素の差のストリームを返します。 ご想像のとおり、最終ストリームでは元の要素より要素が1つ少なくなります。

pairMap通常の中間操作のようpairMap見せる、つまり、実際には、ターミナル操作になるまで計算を実行しないようにします。 これを行うには、入力ストリームからspliteratorを取得する必要がありますが、求められるまで何もしません。 もう1つの小さいが重要なこと： close()新しいスレッドをclose()ときは、元のスレッドを閉じる必要があります。 その結果、メソッドは次のようになります。

 public static <T, R> Stream<R> pairMap(Stream<T> stream, BiFunction<T, T, R> mapper) { return StreamSupport.stream(new PairSpliterator<>(mapper, stream.spliterator()), stream.isParallel()).onClose(stream::close); } 

spliterator()メソッドを呼び出した後のソースストリームは「使用済み」になり、これ以上おを調理することはできません。 しかし、これは正常です。これは、新しい操作を追加するときにすべての中間スレッドで発生します。 2つのストリームを結合するStream.concat（）メソッドは次のようになります。 PairSpliterator自体を記述するPairSpliteratorます。

ポイントに行きましょう。

最も簡単なことは、 characteristics()メソッドを記述することです。 一部の特性は元のスプリッターから継承されますが、NONNULL、DISTINCT、およびSORTEDをリセットする必要があります。任意のマッパー関数を使用した後にこれらの特性を保証することはできません。

 public int characteristics() { return source.characteristics() & (SIZED | SUBSIZED | CONCURRENT | IMMUTABLE | ORDERED); } 

tryAdvanceメソッドの実装は、かなり単純でなければなりません。 元のスプリッターからデータを読み取り、中間バッファーの前の要素を記憶し、最後のペアのmapperを呼び出すだけです。 ストリームの先頭にいるかどうかを覚えておくだけの価値があります。 ブール変数hasPrevこれに役立ち、前の値があるかどうかを示します。

trySplitメソッドtrySplit 、ソーススプリッターでtrySplitを呼び出すことで実装するのtrySplit最適です。 ここでの主な困難は、元のストリームの2つの分離された部分のジャンクションでペアを処理することです。 このペアは、前半をバイパスするスプリッターで処理する必要があります。 したがって、後半の最初の値を保存し、最後に達したときに、最後の値とともにmapperに送信することで再び動作する必要があります。

これに対処した後、コンストラクターを作成します。

 class PairSpliterator<T, R> implements Spliterator<R> { Spliterator<T> source; boolean hasLast, hasPrev; private T cur; private final T last; private final BiFunction<T, T, R> mapper; public PairSpliterator(BiFunction<T, T, R> mapper, Spliterator<T> source) { this(mapper, source, null, false, null, false); } public PairSpliterator(BiFunction<T, T, R> mapper, Spliterator<T> source, T prev, boolean hasPrev, T last, boolean hasLast) { this.source = source; //   this.hasLast = hasLast; //       (   ) this.hasPrev = hasPrev; //     this.cur = prev; //   this.last = last; //     this.mapper = mapper; } // ... } 

tryAdvanceメソッド（ラムダを使用して元のtryAdvanceに渡す代わりに、セッターへのリンクを使用）：

 void setCur(T t) { cur = t; } @Override public boolean tryAdvance(Consumer<? super R> action) { if (!hasPrev) { //    :      if (!source.tryAdvance(this::setCur)) { return false; //    —  } hasPrev = true; } T prev = cur; //    if (!source.tryAdvance(this::setCur)) { //     if (!hasLast) return false; //    —  hasLast = false; //       cur = last; } action.accept(mapper.apply(prev, cur)); //   action  mapper' return true; } 

そして、ここにtrySplit()メソッドがあります：

 public Spliterator<R> trySplit() { Spliterator<T> prefixSource = source.trySplit(); //    if (prefixSource == null) return null; //   —       T prev = cur; //       ,     if (!source.tryAdvance(this::setCur)) { //      source = prefixSource; //      —    return null; } boolean oldHasPrev = hasPrev; hasPrev = true; //      ,      return new PairSpliterator<>(mapper, prefixSource, prev, oldHasPrev, cur, true); } 

estimateSize()を記述することは難しくありません。ソーススプリッターがそのサイズを推定できる場合は、フラグを確認し、1つまたは2つのユニットを微調整するだけです。

 public long estimateSize() { long size = source.estimateSize(); if (size == Long.MAX_VALUE) //       —     return size; if (hasLast) //          size++; if (!hasPrev && size > 0) //        size--; return size; } 

このフォームでは、このスプリッターが私のStreamExライブラリに入りました。 唯一の違いは、プリミティブ型のバージョンを作成する必要があったことです。まあ、 pairMapは静的メソッドではありません。

これはすべて非常に遅いようですか？

すべてが速度でそれほど悪くはありません。 たとえば、StackOverflowの次の問題を考えてみましょう。指定されたIntegerセットから、それらに続く数よりも小さいものだけを残し、結果を新しいリストに保存します。 タスク自体は非常に単純なので、時間のかなりの部分がオーバーヘッドに費やされます。 2つの素朴な実装を提案できます。イテレータ（任意のコレクションで動作します）と要素番号によるアクセス（高速ランダムアクセスのリストでのみ動作します）です。 イテレータ（naiveIterator）を使用したバリアントは次のとおりです。

 List<Integer> result = new ArrayList<>(); Integer last = null; for (Integer cur : input) { if (last != null && last < cur) result.add(last); last = cur; } 

ただし、ランダムアクセス（naiveGet）の場合：

 List<Integer> result = new ArrayList<>(); for (int i = 0; i < input.size() - 1; i++) { Integer cur = input.get(i), next = input.get(i + 1); if (cur < next) result.add(cur); } 

StreamExライブラリを使用したソリューションは非常にコンパクトで、どのデータソース（streamEx）でも機能します。

 List<Integer> result = StreamEx.of(input).pairMap((a, b) -> a < b ? a : null).nonNull().toList(); 

コメンテーターは、さらに3つの実用的なソリューションを提案しました。 従来の多かれ少なかれ投票数が最も多かったため、入力時にランダムアクセスリストが必要です（このソリューションストリームを呼び出しましょう）。

 List<Integer> result = IntStream.range(0, input.size() - 1).filter(i -> input.get(i) < input.get(i + 1)).mapToObj(input::get) .collect(Collectors.toList()); 

以下は、並列化（削減）しない副作用を伴う削減です。

 List<Integer> result = new ArrayList<>(); input.stream().reduce((a, b) -> { if (a < b) result.add(a); return b; }); 

最後の1つはコレクターであり、これも並列ではありません（コレクター）。

 public static Collector<Integer, ?, List<Integer>> collectPrecedingValues() { int[] holder = { Integer.MAX_VALUE }; return Collector.of(ArrayList::new, (l, elem) -> { if (holder[0] < elem) l.add(holder[0]); holder[0] = elem; }, (l1, l2) -> { throw new UnsupportedOperationException("Don't run in parallel"); }); } List<Integer> result = input.stream().collect(collectPrecedingValues()); 

ストリームとstreamExの並列バージョンも比較対象になります。 長さn = 10,000、100,000、1,000,000の要素のランダムな整数の配列で実験を行います（約半分が結果になります）。 完全なJMHベンチマークコードはこちらです。 すべてのアルゴリズムが同じ結果の配列を生成することが検証されます。

測定は、クアッドコアCore-i5で実行されました。 結果は次のようになります（操作ごとにマイクロ秒単位ですべての時間、少ない方が良い）：

アルゴリズム	n = 10,000	n = 100,000	n = 1,000,000
naiveIterator	97.7	904.0	10592.7
ナイーブ	99.8	1084.4	11424.2
コレクター	112.5	1404.9	14387.2
減らす	112.1	1139.5	12001.5
ストリーム	146.4	1624.1	16600.9
streamEx	115.2	1247.1	12967.0
streamParallel	56.9	582.3	6120.5
streamExParallel	53.4	516.7	5353.4

pairMap（streamEx）を含むバージョンは、従来のストリーミングバージョン（stream）とコレクターを含むバージョンの両方を追い越していることがわかります。 同時に、streamExのパラレルバージョンはストリームのパラレルバージョンよりも高速であり、小さなデータセットであってもすべてのシリアルバージョンを大幅に上回ります。 これは、 Stream Parallel Guidanceの経験則と一致しています。少なくとも100マイクロ秒実行される場合、タスクを並列化するのは理にかなっています。

独自のストリームを作成する場合は、適切なスプリッターに応じて、タスクの並列化方法に依存することに注意してください。 分割に煩わされたくない場合は、スプリッターをまったく作成せずに、ユーティリティメソッドを使用してください。 また、既存のJDK機能を使用してストリームを作成できる場合は、新しいスプリッターを作成しないでください。 優れたスプリッターがあれば、それほど難しくないタスクでも並列処理で高速化できます。