正常情况下，一个流在执行一次终端操作之后便结束了。本文通过复制流内数据的方式，曲折的实现了同一个流上执行多次操作。

public class StreamForker
    
      {    private final Stream
     
       stream;    private final Map
      
       , ?>> forks = new HashMap<>();    public StreamForker(Stream
       
         stream) {        this.stream = stream;    }    public StreamForker
        
          fork(Object key, Function
         
          
           , ?> f) { forks.put(key, f); return this; } public Results getResults() { ForkingStreamConsumer
           
             consumer = build(); try { stream.sequential().forEach(consumer); } finally { consumer.finish(); } return consumer; } private ForkingStreamConsumer
            
              build() { List
             
              
               > queues = new ArrayList<>(); Map
               
                > 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
                 
                  
                   > queues, Function
                   
                    
                     , ?> f) { BlockingQueue
                     
                       queue = new LinkedBlockingQueue<>(); queues.add(queue); Spliterator
                      
                        spliterator = new BlockingQueueSpliterator<>(queue); Stream
                       
                         source = StreamSupport.stream(spliterator, false); return CompletableFuture.supplyAsync(() -> f.apply(source)); }}
                       
                      
                     
                    
                   
                  
                 
                
               
              
             
            
           
          
         
        
       
      
     
    

accept方法将原始流中所有的数据添加到各个BlockingQueue内，此处实现了复制

class ForkingStreamConsumer
    
      implements Consumer
     
      , Results {    static final Object END_OF_STREAM = new Object();    private final List
      
       
        > queues;    private final Map
        
         > actions;    public ForkingStreamConsumer(List
         
          
           > queues, Map
           
            > actions) { this.queues = queues; this.actions = actions; } @Override public void accept(T t) { queues.forEach(q -> q.add(t)); } @SuppressWarnings("unchecked") void finish() { accept((T) END_OF_STREAM); } @SuppressWarnings("unchecked") @Override public 
            
              R get(Object key) { try { return ((Future
             
              ) actions.get(key)).get(); } catch (Exception e) { throw new RuntimeException(e); } }}
             
            
           
          
         
        
       
      
     
    

此处重写了tryAdvance接口，只是简单的从BlockingQueue中取出数据，执行action。业务逻辑中复制流是为了做什么事情，action就是这件事情。ForkingStreamConsumer.END_OF_STREAM是Queue中数据结束的标示

class BlockingQueueSpliterator
    
      implements Spliterator
     
       {    private final BlockingQueue
      
        q;    BlockingQueueSpliterator(BlockingQueue
       
         q) {        this.q = q;    }    @Override    public boolean tryAdvance(Consumer
         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;    }    @Override    public Spliterator
        
          trySplit() {        return null;    }    @Override    public long estimateSize() {        return 0;    }    @Override    public int characteristics() {        return 0;    }}