ストリームのreduce（）操作を短絡する方法はありますか？

Question

これは本質的にHow to short-circuit reduce on Stream?と同じ質問です。しかし、この質問はブール値のストリームに焦点を当てており、その答えは他の型に対して一般化できず、操作を減らすことができないので、より一般的な質問をしたいと思います。

還元操作のためにabsorbing elementに遭遇したとき、ストリームを短絡させて短絡させるにはどうすればよいですか？

典型的な数学的ケースは、乗算のために0になります。このStream：

int product = IntStream.of(2, 3, 4, 5, 0, 7, 8) 
     .reduce(1, (a, b) -> a * b); 

かかわらず0に遭遇された後、製品が知られているという事実の最後の二つの要素（7と8）を消費します。

Answer 1

残念ながら、Stream APIには、独自の短絡操作を作成する機能がありません。あまりにもきれいな解決策は、RuntimeExceptionを投げてキャッチすることです。ここでIntStreamの実装ですが、それは、他のストリームのタイプのために一般化することができます。

public static int reduceWithCancelEx(IntStream stream, int identity, 
         IntBinaryOperator combiner, IntPredicate cancelCondition) { 
    class CancelException extends RuntimeException { 
     private final int val; 

     CancelException(int val) { 
      this.val = val; 
     } 
    } 

    try { 
     return stream.reduce(identity, (a, b) -> { 
      int res = combiner.applyAsInt(a, b); 
      if(cancelCondition.test(res)) 
       throw new CancelException(res); 
      return res; 
     }); 
    } catch (CancelException e) { 
     return e.val; 
    } 
} 

使用例：

int product = reduceWithCancelEx(
     IntStream.of(2, 3, 4, 5, 0, 7, 8).peek(System.out::println), 
     1, (a, b) -> a * b, val -> val == 0); 
System.out.println("Result: "+product); 

出力：それは作品にもかかわらず、

2 
3 
4 
5 
0 
Result: 0 

注意並列ストリームでは、他の並列タスクが例外をスローするとすぐに終了することは保証されません。すでに開始されているサブタスクは完了するまで実行される可能性が高いため、予想以上に多くの要素を処理する可能性があります。

更新：はるかに長いが、より並列的な代替ソリューションです。これは、基本的なスプライテータに基づいています。このスプライテータは、基本要素をすべて累積した結果である最大でも1つの要素を返します。シーケンシャルモードで使用すると、すべての作業が単一のtryAdvanceコールで実行されます。分割すると、各パートは対応する単一の部分結果を生成し、コンバイナ関数を使用してストリームエンジンによって削減されます。ここには一般的なバージョンがありますが、原始的な特殊化も可能です。 Stream.reduce(identity, accumulator, combiner)とStream.reduce(identity, combiner)に似てい

final static class CancellableReduceSpliterator<T, A> implements Spliterator<A>, 
     Consumer<T>, Cloneable { 
    private Spliterator<T> source; 
    private final BiFunction<A, ? super T, A> accumulator; 
    private final Predicate<A> cancelPredicate; 
    private final AtomicBoolean cancelled = new AtomicBoolean(); 
    private A acc; 

    CancellableReduceSpliterator(Spliterator<T> source, A identity, 
      BiFunction<A, ? super T, A> accumulator, Predicate<A> cancelPredicate) { 
     this.source = source; 
     this.acc = identity; 
     this.accumulator = accumulator; 
     this.cancelPredicate = cancelPredicate; 
    } 

    @Override 
    public boolean tryAdvance(Consumer<? super A> action) { 
     if (source == null || cancelled.get()) { 
      source = null; 
      return false; 
     } 
     while (!cancelled.get() && source.tryAdvance(this)) { 
      if (cancelPredicate.test(acc)) { 
       cancelled.set(true); 
       break; 
      } 
     } 
     source = null; 
     action.accept(acc); 
     return true; 
    } 

    @Override 
    public void forEachRemaining(Consumer<? super A> action) { 
     tryAdvance(action); 
    } 

    @Override 
    public Spliterator<A> trySplit() { 
     if(source == null || cancelled.get()) { 
      source = null; 
      return null; 
     } 
     Spliterator<T> prefix = source.trySplit(); 
     if (prefix == null) 
      return null; 
     try { 
      @SuppressWarnings("unchecked") 
      CancellableReduceSpliterator<T, A> result = 
       (CancellableReduceSpliterator<T, A>) this.clone(); 
      result.source = prefix; 
      return result; 
     } catch (CloneNotSupportedException e) { 
      throw new InternalError(); 
     } 
    } 

    @Override 
    public long estimateSize() { 
     // let's pretend we have the same number of elements 
     // as the source, so the pipeline engine parallelize it in the same way 
     return source == null ? 0 : source.estimateSize(); 
    } 

    @Override 
    public int characteristics() { 
     return source == null ? SIZED : source.characteristics() & ORDERED; 
    } 

    @Override 
    public void accept(T t) { 
     this.acc = accumulator.apply(this.acc, t); 
    } 
} 

方法が、cancelPredicateと：

public static <T, U> U reduceWithCancel(Stream<T> stream, U identity, 
     BiFunction<U, ? super T, U> accumulator, BinaryOperator<U> combiner, 
     Predicate<U> cancelPredicate) { 
    return StreamSupport 
      .stream(new CancellableReduceSpliterator<>(stream.spliterator(), identity, 
        accumulator, cancelPredicate), stream.isParallel()).reduce(combiner) 
      .orElse(identity); 
} 

public static <T> T reduceWithCancel(Stream<T> stream, T identity, 
     BinaryOperator<T> combiner, Predicate<T> cancelPredicate) { 
    return reduceWithCancel(stream, identity, combiner, combiner, cancelPredicate); 
} 

のは、両方のバージョンをテストし、実際に処理されているどのように多くの要素を数えてみましょう。 0を近づけてみましょう。例外のバージョン：

AtomicInteger count = new AtomicInteger(); 
int product = reduceWithCancelEx(
     IntStream.range(-1000000, 100).filter(x -> x == 0 || x % 2 != 0) 
       .parallel().peek(i -> count.incrementAndGet()), 1, 
     (a, b) -> a * b, x -> x == 0); 
System.out.println("product: " + product + "/count: " + count); 
Thread.sleep(1000); 
System.out.println("product: " + product + "/count: " + count); 

典型的な出力：唯一のいくつかの要素が処理されるとき、結果が返されながら

product: 0/count: 281721 
product: 0/count: 500001 

ので、タスクはバックグラウンドで作業し、カウンタがまだ増加している続けています。

AtomicInteger count = new AtomicInteger(); 
int product = reduceWithCancel(
     IntStream.range(-1000000, 100).filter(x -> x == 0 || x % 2 != 0) 
       .parallel().peek(i -> count.incrementAndGet()).boxed(), 
       1, (a, b) -> a * b, x -> x == 0); 
System.out.println("product: " + product + "/count: " + count); 
Thread.sleep(1000); 
System.out.println("product: " + product + "/count: " + count); 

典型的な出力：ここspliteratorバージョンです

product: 0/count: 281353 
product: 0/count: 281353 

結果が返されたときにすべてのタスクが実際に終了します。

Answer 2

一般的な短絡静的低減方法は、ストリームのスプライテータを使用して実装できます。それはあまり複雑ではないことが判明しました！ spliteratorの使用は、より柔軟な方法でSteamを使用したい場合に、多くの時間を費やす方法と思われます。

public static <T> T reduceWithCancel(Stream<T> s, T acc, BinaryOperator<T> op, Predicate<? super T> cancelPred) { 
    BoxConsumer<T> box = new BoxConsumer<T>(); 
    Spliterator<T> splitr = s.spliterator(); 

    while (!cancelPred.test(acc) && splitr.tryAdvance(box)) { 
     acc = op.apply(acc, box.value); 
    } 

    return acc; 
} 

public static class BoxConsumer<T> implements Consumer<T> { 
    T value = null; 
    public void accept(T t) { 
     value = t; 
    } 
} 

使用：

int product = reduceWithCancel(
     Stream.of(1, 2, 0, 3, 4).peek(System.out::println), 
     1, (acc, i) -> acc * i, i -> i == 0); 

    System.out.println("Result: " + product); 

出力：

1 
2 
0 
Result: 0 

方法は、端末の動作の他の種類を実行するように一般化することができます。

これはtake-while操作についてのthis answerに緩く基づいています。

私はこの並列化の可能性について何も知らない。

Answer 3

これは自分ではreduce()を使用しないで、既存の短絡の最終動作を使用することです。

副作用のある述語を使用する場合、noneMatch（）またはallMatch（）を使用してこれを使用できます。確かに最も清潔な解決策ではありませんが、目標を達成します。

AtomicInteger product = new AtomicInteger(1); 
IntStream.of(2, 3, 4, 5, 0, 7, 8) 
     .peek(System.out::println) 
     .noneMatch(i -> { 
      if (i == 0) { 
       product.set(0); 
       return true; 
      } 
      int oldValue = product.get(); 
      while (oldValue != 0 && !product.compareAndSet(oldValue, i * oldValue)) { 
       oldValue = product.get(); 
      } 
      return oldValue == 0; 
     }); 
System.out.println("Result: " + product.get()); 

短絡して並列にすることができます。