Java 8 如何以并发方式在同一个流上执行多种操作

Java 8 中，流有一个非常大的（也可能是最大的）局限性，使用时，对它操作一次仅能得到一个处理结果。实际操作中，如果你试图多次遍历同一个流，结果只有一个，那就是遭遇下面这样的异常：
java.lang.IllegalStateException: stream has already been operated upon or closed
虽然流的设计就是如此，但我们在处理流时经常希望能同时获取多个结果。
本篇利用一个通用API，即Spliterator，尤其是它的延迟绑定能力，结合BlockingQueues和Futures来实现这一大有裨益的特性。

1.复制流

要达到在一个流上并发地执行多个操作的效果，你需要做的第一件事就是创建一个StreamForker，这个StreamForker会对原始的流进行封装，在此基础之上你可以继续定义你希望执行的各种操作。我们看看下面这段代码。

public class StreamForker<T> {private final Stream<T> stream;private final Map<Object, Function<Stream<T>, ?>> forks = new HashMap<>();public StreamForker(Stream<T> stream) {this.stream = stream;}/*** 这里的fork方法接受两个参数。* Function参数，它对流进行处理，将流转变为代表这些操作结果的任何类型。* key参数，通过它你可以取得操作的结果，并将这些键/函数对累积到一个内部的Map中。** @param key* @param f* @return*/public StreamForker<T> fork(Object key, Function<Stream<T>, ?> f) {forks.put(key, f);return this; //返回this从而保证多次流畅地调用fork方法}public Results getResults() {ForkingStreamConsumer<T> consumer = build();try {stream.sequential().forEach(consumer);} finally {consumer.finish();}return consumer;}}

所有由fork方法添加的操作的执行都是通过getResults方法的调用触发的，该方法返回一个Results接口的实现，具体的定义如下：

 public interface Results {<R> R get(Object key);
}

1.1 使用 ForkingStreamConsumer 实现 Results 接口

你可以用下面的方式实现getResults方法：

 public Results getResults() {ForkingStreamConsumer<T> consumer = build();try {stream.sequential().forEach(consumer);} finally {consumer.finish();}return consumer;}

ForkingStreamConsumer同时实现了前面定义的Results接口和Consumer接口。随着我们进一步剖析它的实现细节，你会看到它主要的任务就是处理流中的元素，将它们分发到多个BlockingQueues中处理，BlockingQueues的数量和通过fork方法提交的操作数是一致的。注意，我们很明确地知道流是顺序处理的，不过，如果你在一个并发流上执行forEach方法，它的元素可能就不是顺序地被插入到队列中了。finish方法会在队列的末尾插入特殊元素表明该队列已经没有更多需要处理的元素了。build方法主要用于创建ForkingStreamConsumer。

 private ForkingStreamConsumer<T> build() {//创建由队列组成的列表，每一个队列对应一个操作List<BlockingQueue<T>> queues = new ArrayList<>();//建立用于标识操作的键与包含操作结果的Future之间的映射关系HashMap<Object, Future<?>> actions = forks.entrySet().stream().reduce(new HashMap<>(),(map, e) -> {map.put(e.getKey(), getOperationResult(queues, e.getValue()));return map;},(m1, m2) -> {m1.putAll(m2);return m1;});return new ForkingStreamConsumer<>(queues, actions);}

首先创建了我们前面提到的由BlockingQueues组成的列表。紧接着，你创建了一个Map，Map的键就是你在流中用于标识不同操作的键，值包含在Future中，Future中包含了这些操作对应的处理结果。BlockingQueues的列表和Future组成的Map会被传递给ForkingStreamConsumer的构造函数。每个Future都是通过getOperationResult方法创建。

 private Future<?> getOperationResult(List<BlockingQueue<T>> queues, Function<Stream<T>, ?> f) {BlockingQueue<T> queue = new LinkedBlockingDeque<>();queues.add(queue);//创建一个队列并将其添加到队列的列表中Spliterator<T> spliterator = new BlockingQueueSpliterator<>(queue);//创建一个流，将Spliterator作为数据源Stream<T> source = StreamSupport.stream(spliterator, false);//创建一个Future对象，以异步方式计算在流上执行特定函数的结果return CompletableFuture.supplyAsync(() -> f.apply(source));}

getOperationResult方法会创建一个新的BlockingQueue，并将其添加到队列的列表。这个队列会被传递给一个新的BlockingQueueSpliterator对象，后者是一个延迟绑定的
Spliterator，它会遍历读取队列中的每个元素。
接下来创建了一个顺序流对该Spliterator进行遍历，最终会创建一个Future在流上执行某个希望的操作并收集其结果。这里的Future使用CompletableFuture类的一个静态工厂方法创建，CompletableFuture实现了Future接口。

1.2 开发 ForkingStreamConsumer 和 BlockingQueueSpliterator

 static class ForkingStreamConsumer<T> implements Consumer<T>, Results {static final Object END_OF_STREAM = new Object();private final List<BlockingQueue<T>> queues;private final Map<Object, Future<?>> actions;ForkingStreamConsumer(List<BlockingQueue<T>> queues, Map<Object, Future<?>> actions) {this.queues = queues;this.actions = actions;}@Overridepublic void accept(T t) {queues.forEach(q -> q.add(t));//将流中遍历的元素添加到所有的队列中}/*** 将最后一个元素添加到队列中，* 表明该流已经结束*/void finish() {accept((T) END_OF_STREAM);}/*** 等待futuire完成相关的计算，返回由特定键标识的处理结果** @param key* @param <R>* @return*/@Overridepublic <R> R get(Object key) {try {return ((Future<R>) actions.get(key)).get();} catch (Exception e) {throw new RuntimeException(e);}}}

这个类同时实现了Consumer和Results接口，并持有两个引用，一个指向由BlockingQueues组成的列表，另一个是执行了由Future构成的Map结构，它们表示的是即将在流上执行的各种操作。
    Consumer接口要求实现accept方法。这里，每当ForkingStreamConsumer接受流中的一个元素，它就会将该元素添加到所有的BlockingQueues中。另外，当原始流中的所有元素都添
加到所有队列后，finish方法会将最后一个元素添加所有队列。BlockingQueueSpliterators碰到最后这个元素时会知道队列中不再有需要处理的元素了。
    Results接口需要实现get方法。一旦处理结束，get方法会获得Map中由键索引的Future，解析处理的结果并返回。
    最后，流上要进行的每个操作都会对应一个BlockingQueueSpliterator。每个BlockingQueueSpliterator都持有一个指向BlockingQueues的引用，这个BlockingQueues是由ForkingStreamConsumer 生成的。

/*** 一个遍历BlockingQueue并读取其中元素的Spliterator* @param <T>*/
class BlockingQueueSpliterator<T> implements Spliterator<T> {private final BlockingQueue<T> q;BlockingQueueSpliterator(BlockingQueue<T> q) {this.q = q;}@Overridepublic boolean tryAdvance(Consumer<? super T> action) {T t;while (true) {try {t = q.take();break;} catch (InterruptedException e) {}}if (t != ForkingStreamConsumer.END_OF_STREAM) {action.accept(t);return true;}return false;}@Overridepublic Spliterator<T> trySplit() {return null;}@Overridepublic long estimateSize() {return 0;}@Overridepublic int characteristics() {return 0;}}

这段代码实现了一个Spliterator，不过它并未定义如何切分流的策略，仅仅利用了流的延迟绑定能力。由于这个原因，它也没有实现trySplit方法。

由于无法预测能从队列中取得多少个元素，所以estimatedSize方法也无法返回任何有意义的值。更进一步，由于你没有试图进行任何切分，所以这时的估算也没什么用处。

这一实现并没有体现Spliterator的任何特性，因此characteristic方法返回0。

这段代码中提供了实现的唯一方法是tryAdvance，它从BlockingQueue中取得原始流中的元素，而这些元素最初由ForkingSteamConsumer添加。依据getOperationResult方法创建Spliterator同样的方式，这些元素会被作为进一步处理流的源头传递给Consumer对象（在流上要执行的函数会作为参数传递给某个fork方法调用）。tryAdvance方法返回true通知调用方还有其他的元素需要处理，直到它发现由ForkingSteamConsumer添加的特殊对象，表明队列中已经没有更多需要处理的元素了。

2. 所有代码

public class StreamForker<T> {private final Stream<T> stream;private final Map<Object, Function<Stream<T>, ?>> forks = new HashMap<>();public StreamForker(Stream<T> stream) {this.stream = stream;}/*** 这里的fork方法接受两个参数。* Function参数，它对流进行处理，将流转变为代表这些操作结果的任何类型。* key参数，通过它你可以取得操作的结果，并将这些键/函数对累积到一个内部的Map中。** @param key* @param f* @return*/public StreamForker<T> fork(Object key, Function<Stream<T>, ?> f) {forks.put(key, f);return this; //返回this从而保证多次流畅地调用fork方法}public Results getResults() {ForkingStreamConsumer<T> consumer = build();try {stream.sequential().forEach(consumer);} finally {consumer.finish();}return consumer;}private ForkingStreamConsumer<T> build() {//创建由队列组成的列表，每一个队列对应一个操作List<BlockingQueue<T>> queues = new ArrayList<>();//建立用于标识操作的键与包含操作结果的Future之间的映射关系HashMap<Object, Future<?>> actions = forks.entrySet().stream().reduce(new HashMap<>(),(map, e) -> {map.put(e.getKey(), getOperationResult(queues, e.getValue()));return map;},(m1, m2) -> {m1.putAll(m2);return m1;});return new ForkingStreamConsumer<>(queues, actions);}private Future<?> getOperationResult(List<BlockingQueue<T>> queues, Function<Stream<T>, ?> f) {BlockingQueue<T> queue = new LinkedBlockingDeque<>();queues.add(queue);//创建一个队列并将其添加到队列的列表中Spliterator<T> spliterator = new BlockingQueueSpliterator<>(queue);//创建一个流，将Spliterator作为数据源Stream<T> source = StreamSupport.stream(spliterator, false);//创建一个Future对象，以异步方式计算在流上执行特定函数的结果return CompletableFuture.supplyAsync(() -> f.apply(source));}static class ForkingStreamConsumer<T> implements Consumer<T>, Results {static final Object END_OF_STREAM = new Object();private final List<BlockingQueue<T>> queues;private final Map<Object, Future<?>> actions;ForkingStreamConsumer(List<BlockingQueue<T>> queues, Map<Object, Future<?>> actions) {this.queues = queues;this.actions = actions;}@Overridepublic void accept(T t) {queues.forEach(q -> q.add(t));//将流中遍历的元素添加到所有的队列中}/*** 将最后一个元素添加到队列中，* 表明该流已经结束*/void finish() {accept((T) END_OF_STREAM);}/*** 等待futuire完成相关的计算，返回由特定键标识的处理结果** @param key* @param <R>* @return*/@Overridepublic <R> R get(Object key) {try {return ((Future<R>) actions.get(key)).get();} catch (Exception e) {throw new RuntimeException(e);}}}/*** 一个遍历BlockingQueue并读取其中元素的Spliterator* @param <T>*/class BlockingQueueSpliterator<T> implements Spliterator<T> {private final BlockingQueue<T> q;BlockingQueueSpliterator(BlockingQueue<T> q) {this.q = q;}@Overridepublic boolean tryAdvance(Consumer<? super T> action) {T t;while (true) {try {t = q.take();break;} catch (InterruptedException e) {}}if (t != ForkingStreamConsumer.END_OF_STREAM) {action.accept(t);return true;}return false;}@Overridepublic Spliterator<T> trySplit() {return null;}@Overridepublic long estimateSize() {return 0;}@Overridepublic int characteristics() {return 0;}}interface Results {<R> R get(Object key);}public static void main(String[] args) {//测试List<Integer> menu = Arrays.asList(1, 2, 3, 4, 5, 6, 20, 40, 60);Results results = new StreamForker<Integer>(menu.stream()).fork("max", s -> s.max(Integer::compareTo)).fork("sum", s -> s.collect(Collectors.summarizingInt(Integer::intValue))).getResults();System.out.println("max:"+results.get("max"));System.out.println("sum:"+results.get("sum"));}
}

3. 性能的考量

提起性能，你不应该想当然地认为这种方法比多次遍历流的方式更加高效。如果构成流的数据都保存在内存中，阻塞式队列所引发的开销很容易就抵消了由并发执行操作所带来的性能
提升。

与此相反，如果操作涉及大量的I/O，譬如流的源头是一个巨型文件，那么单次访问流可能是个不错的选择；因此（大多数情况下）优化应用性能唯一有意义的规则是“好好地度量它”。

通过这个例子，我们展示了怎样一次性地在同一个流上执行多个操作。更重要地是，我们相信这个例子也证明了一点，即使某个特性原生的Java API暂时还不支持，充分利用Lambda表达式的灵活性和一点点的创意，整合现有的功能，你完全可以实现想要的新特性。