如何在Java 8中爲Stream API實現並行支持takeWhile？

Question

我已經看到了一些用於Java 8流API的takeWhile實現，但它們似乎都將流轉換爲非並行流。例如this之一：

static <T> Spliterator<T> takeWhile(
    Spliterator<T> splitr, Predicate<? super T> predicate) { 
    return new Spliterators.AbstractSpliterator<T>(splitr.estimateSize(), 0) { 
    boolean stillGoing = true; 
    @Override public boolean tryAdvance(Consumer<? super T> consumer) { 
     if (stillGoing) { 
     boolean hadNext = splitr.tryAdvance(elem -> { 
      if (predicate.test(elem)) { 
      consumer.accept(elem); 
      } else { 
      stillGoing = false; 
      } 
     }); 
     return hadNext && stillGoing; 
     } 
     return false; 
    } 
    }; 
} 

static <T> Stream<T> takeWhile(Stream<T> stream, Predicate<? super T> predicate) { 
    return StreamSupport.stream(takeWhile(stream.spliterator(), predicate), false); 
} 

這裏StreamSupport.stream(takeWhile(stream.spliterator(), predicate), false);轉動傳遞給takeWhile成連續流的流。是否有人知道支持並行流的實現，或者我如何修改此代碼以使其維護/支持並行流？

Answer 1

如果源被稱爲是無序的，那麼下面的實現應該工作：

static <T> Stream<T> takeWhile(Stream<T> stream, Predicate<? super T> predicate) { 
    return StreamSupport.stream(UnorderedTakeWhileSpliterator<>(stream.spliterator(), predicate), stream.isParallel()); 
} 

有序實施將作爲更棘手：

static final class UnorderedTakeWhileSpliterator<T> implements Spliterator<T>, Consumer<T>, Cloneable { 
    private final Predicate<? super T> predicate; 
    private final AtomicBoolean checked = new AtomicBoolean(); 
    private Spliterator<T> source; 
    private T cur; 

    UnorderedTakeWhileSpliterator(Spliterator<T> source, Predicate<? super T> predicate) { 
     this.predicate = predicate; 
     this.source = source; 
    } 

    @Override 
    public void accept(T t) { 
     this.cur = t; 
    } 

    @Override 
    public boolean tryAdvance(Consumer<? super T> action) { 
     if (!checked.get() && source.tryAdvance(this)) { 
      if (predicate.test(cur)) { 
       action.accept(cur); 
       return true; 
      } else { 
       checked.set(true); 
      } 
     } 
     return false; 
    } 

    @Override 
    public Spliterator<T> trySplit() { 
     Spliterator<T> prefix = source.trySplit(); 
     if(prefix == null) { 
      return null; 
     } 
     if(checked.get()) { 
      return Spliterators.emptySpliterator(); 
     } 
     UnorderedTakeWhileSpliterator<T> clone; 
     try { 
      clone = (UnorderedTakeWhileSpliterator<T>) clone(); 
     } catch (CloneNotSupportedException e) { 
      throw new InternalError(e); 
     } 
     clone.source = prefix; 
     return clone; 
    } 

    @Override 
    public long estimateSize() { 
     return source.estimateSize(); 
    } 

    @Override 
    public int characteristics() { 
     return source.characteristics() & (DISTINCT | SORTED | NONNULL); 
    } 

    @Override 
    public Comparator<? super T> getComparator() { 
     return source.getComparator(); 
    } 
} 

以下方法創建流它應該緩衝非前綴項並將取消傳播到後綴。類似這樣的東西在JDK-9中實現（不是作爲分割器，而是作爲普通的流操作），但我懷疑即使是這種棘手的實現在很多情況下都勝過順序流。