ギャザラは、入力要素のストリームを出力要素のストリームに変換する中間操作であり、入力要素のストリームの最後に到達したときに、オプションで最終アクションを適用します。

Stream.map(Function)などの中間操作では新しいストリームが生成され、Stream.forEach(Consumer)などの終端操作ではストリーム以外の結果が生成されることに注意してください。ストリーム以外の結果は、プリミティブ値(double値など)やコレクションのこともあれば、forEachの場合など、値がまったく存在しないこともあります。

Gatherersは、次のことを実行できます。

• 要素を1対1、1対多、多対1、多対多の方法で変換します。
• 以前に確認した要素を追跡して、後の要素の変換に影響を与えます。
• 短絡、または入力要素の処理を停止して、無限ストリームを有限ストリームに変換します
• ストリームをパラレルで処理します

  ノート:

  収集では、ギャザラの作成時にコンバイナ関数を指定した場合にのみ、ストリームがパラレルで処理されます。「ギャザラの作成」の「コンバイナ関数」を参照してください。parallel()をコールした場合でも、ギャザラのデフォルト・コンバイナによってパラレル化がオフになります。

収集操作の例を次に示します。

• 要素のバッチへのグループ化
• 連続した類似要素の重複除外
• 増分累積関数
• 増分順序変更関数

ギャザラを作成するには、Gathererインタフェースを実装します。

次の例では、整数のストリームから最大の整数を返すギャザラを作成します。ただし、ギャザラがその引数limit以上の整数を検出した場合は、その整数を返し、ストリームの整数の処理を停止します。

    record BiggestInt(int limit) implements Gatherer<Integer, List<Integer>, Integer> {

        // The initializer creates a new private ArrayList to keep track of the
        // largest integer across elements.
        
        @Override
        public Supplier<List<Integer>> initializer() {
            return () -> new ArrayList<Integer>(1);
        }
        
        // The integrator 
       
        @Override
        public Integrator<List<Integer>, Integer, Integer> integrator() {
            return Integrator.of(
                (max, element, downstream) -> {
                    
                    // Save the integer if it's the largest so far.
                    if (max.isEmpty()) max.addFirst(element);
                    else if (element > max.getFirst()) max.set(0, element);
                
                    // If the integer is equal or greater to the limit,
                    // "short-circuit": emit the current integer downstream
                    // and return false to stop processing stream elements
                    if (element >= limit) {
                        downstream.push(element);
                        return false;
                    }
                    
                    // Return true to continue processing stream elements
                    return true;
                }
            );
        }
        
        // The combiner, which is used during parallel evaluation
        
        @Override
        public BinaryOperator<List<Integer>> combiner() {
            return (leftMax, rightMax) -> {
                
                // If either the "left" or "right" ArrayLists contain
                // no value, then return the other
                if (leftMax.isEmpty()) return rightMax;
                if (rightMax.isEmpty()) return leftMax;
                
                // Return the ArrayList that contains the larger integer
                int leftVal = leftMax.getFirst();
                int rightVal = rightMax.getFirst();
                if (leftVal > rightVal) return leftMax;
                else return rightMax;
            };
        }
        
        @Override
        public BiConsumer<List<Integer>, Downstream<? super Integer>> finisher() {
            
            // Emit the largest integer, if there is one, downstream
            return (max, downstream) -> {
                if (!max.isEmpty()) {
                    downstream.push(max.getFirst());
                }
            };
        }
    }

この収集は、次のように使用できます。

    System.out.println(Stream.of(5,4,2,1,6,12,8,9)
                             .gather(new BiggestInt(11))
                             .findFirst()
                             .get());

次のように出力されます。

12

このギャザラをパラレルで使用することもできます。

    System.out.println(Stream.of(5,4,2,1,6,12,8,9)
                             .gather(new BiggestInt(11))
                             .parallel()
                             .findFirst()
                             .get());

Gatherer<T,A,R>インタフェースには、次の3つの型パラメータがあります。

• T: 収集操作の入力要素の型。この例では、Integer要素のストリームを処理します。
• A: ギャザラの非公開状態オブジェクトの型。ギャザラは、以前に確認した要素を追跡して、後の要素の変換に影響を与えるために使用できます。この例では、List<Integer>を使用して、これまでに検出した最大のIntegerを入力ストリームに格納します。
• R: 収集操作からの出力要素の型。この例では、Integer値を返します。

ギャザラを作成するには、入力要素を処理する連携した4つの関数を定義します。これらの関数の一部は、ギャザラの操作に応じてオプションとなります。

• initializer(): ギャザラの非公開状態オブジェクトを作成します
• integrator(): 入力ストリームから新しい要素を統合し、場合によっては非公開状態オブジェクトを検査して、出力ストリームに要素を発行します。
• combiner(): ギャザラがストリームをパラレルで処理しているときに、2つの非公開状態オブジェクトを1つに結合します。
• finisher(): オプションで、ギャザラがすべての入力要素を処理した後、アクションを実行します。これにより、非公開状態オブジェクトを検査したり、追加の出力要素を発行できます。

        @Override
        public Supplier<List<Integer>> initializer() {
            return () -> new ArrayList<Integer>(1);
        }

 (max, element, downstream) -> {
                    
                    // Save the integer if it's the largest so far.
                    if (max.isEmpty()) max.addFirst(element);
                    else if (element > max.getFirst()) max.set(0, element);
                
                    // If the integer is equal or greater to the limit,
                    // "short-circuit": emit the current integer downstream
                    // and return false to stop processing stream elements
                    if (element >= limit) {
                        downstream.push(element);
                        return false;
                    }
                    
                    // Return true to continue processing stream elements
                    return true;
                }

        @Override
        public BinaryOperator<List<Integer>> combiner() {
            return (leftMax, rightMax) -> {
                
                // If either the "left" or "right" ArrayLists contain
                // no value, then return the other
                if (leftMax.isEmpty()) return rightMax;
                if (rightMax.isEmpty()) return leftMax;
                
                // Return the ArrayList that contains the larger integer
                int leftVal = leftMax.getFirst();
                int rightVal = rightMax.getFirst();
                if (leftVal > rightVal) return leftMax;
                else return rightMax;
            };
        }

        @Override
        public BiConsumer<List<Integer>, Downstream<? super Integer>> finisher() {
            
            // Emit the largest integer, if there is one, downstream
            return (max, downstream) -> {
                if (!max.isEmpty()) {
                    downstream.push(max.getFirst());
                }
            };
        }

Gathererインタフェースを実装するかわりに、Gathererインタフェースでファクトリ・メソッドの1つをコールして、ギャザラを作成できます。

次の例は、Gatherer::ofメソッドをコールする点を除き、「ギャザラの作成」で説明した例と同じです。

    static Gatherer<Integer, List<Integer>, Integer> biggestInt(int limit) {
    
        return Gatherer.of(
        
            // Supplier
            
            () -> { return new ArrayList<Integer>(1); },
            
            // Integrator
            
            Gatherer.Integrator.of(
                (max, element, downstream) -> {
                    System.out.println("Processing " + element);
                    if (max.isEmpty()) max.addFirst(element);
                    else if (element > max.getFirst()) max.set(0, element);
                
                    if (element >= limit) {
                        downstream.push(element);
                        return false;
                    }
                    return true;
                }            
            ),
            
            // Combiner
            
            (leftMax, rightMax) -> {
                if (leftMax.isEmpty()) return rightMax;
                if (rightMax.isEmpty()) return leftMax;
                int leftVal = leftMax.getFirst();
                int rightVal = rightMax.getFirst();
                if (leftVal > rightVal) return leftMax;
                else return rightMax;
            },
            
            // Finisher
            
            (max, downstream) -> {
                if (!max.isEmpty()) {
                    downstream.push(max.getFirst());
                }
            }
        );
    }

このギャザラは、次のようにコールできます。

    System.out.println(Stream.of(5,4,2,1,6,12,8,9)
                             .gather(biggestInt(11))
                             .parallel()
                             .findFirst()
                             .get());