Java8 - 处理 Stream> 并行交付给非线程安全消费者的惯用方式？

Question

假设我有一个 Stream<Callable<SomeClass>> stream; .该流正在访问超过一百万个无法放入内存的对象。

将其转换为 Stream<SomeClass> 的惯用方法是什么？以确保 Callable::call 的方式在交付给非线程安全的消费者之前并行执行(可能通过调用 .sequential().forEach() 或其他一些瓶颈机制)？

即并行处理流但按顺序传递输出(随机顺序可以，只要它是单线程的)。

我知道我可以通过设置 ExecutionService 来做我想做的事和一个 Queue在原始流和消费者之间。但这似乎有很多代码，是否有神奇的单行代码？

Answer 1

您仍然可以雇用 ExecutorService用于并行化。像这样:

ExecutorService service = Executors.newFixedThreadPool(4);
stream.map(c -> service.submit(c)).map(future -> {
    try {               
        return future.get(); //retrieve callable result
    } catch (InterruptedException | ExecutionException ex) {        
        //Exception handling    
        throw new RuntimeException(ex);         
    }
});

您可以处理结果 Stream<SomeClass>进一步依次。

如果您直接在 Stream<Future<SomeClass>> 上使用 forEach/forEachOrdered您可以处理结果 SomeClass -一旦当前的 future 完成就直接对象(不同于你使用 invokeAll() 时阻塞，直到每个任务完成)。

如果您想按照可用的确切顺序处理可调用结果，您将不得不使用 CompletionService由于必须调用 Future<SomeClass> f = completionService.take()，因此不能与单个流操作链一起使用提交调用后。

编辑:

使用 ExecutorService在流中并不像我上面显示的那样工作，因为每个 Callable通过 future.get() 提交和请求一个接一个地。

我发现一个可能的甚至副作用更重的解决方案划分 Callables在固定的并行 block 中。

我使用类 TaskMapper作为提交 Callables 的映射函数并将它们映射到 block :

class TaskMapper implements Function<Callable<Integer>, List<Future<Integer>>>{
    private final ExecutorService service;
    private final int chunkSize;
    private List<Future<Integer>> chunk = new ArrayList<>(); 

    TaskMapper(ExecutorService service, int chunkSize){
        this.service = service;
        this.chunkSize = chunkSize;
    }

    @Override
    public List<Future<Integer>> apply(Callable<Integer> c) {
        chunk.add(service.submit(c));
        if(chunk.size() == chunkSize){
            List<Future<Integer>> fList = chunk;
            chunk = new ArrayList<>();              
            return fList;
        }else{
            return null;
        }
    }

    List<Future<Integer>> getChunk(){
        return chunk;
    }
}

这是流操作链的样子:

ExecutorService service = Executors.newFixedThreadPool(4);
TaskMapper taskMapper = new TaskMapper(service, 4);
stream.map(taskMapper)
    .filter(fl -> fl != null) //filter for the chunks
    .flatMap(fl -> fl.stream()) //flat-map the chunks to futures
    .map(future -> {
        try {               
            return future.get();
        } catch (InterruptedException | ExecutionException ex) {    
            throw new RuntimeException(ex);
        }
    });  
//process the remaining futures  
for(Future<Integer> f : taskMapper.getChunk()){
    try {               
        Integer i = f.get();
        //process i
    } catch (InterruptedException | ExecutionException ex) {    
        //exception handling
    }
}

其工作方式如下: TaskMapper每次将 4 个可调用项提交给服务并将它们映射到一大块 future (没有 Spliterator )。这可以通过映射到 null 来解决。每次调用第 1、第 2 和第 3 次。 null例如，可以用虚拟对象替换。将 future 映射到结果的映射函数等待 block 的每个 future 的结果。我用 Integer在我的示例中而不是 SomeClass .当当前 block 中的所有 future 结果都被映射时，将创建一个新的 block 并并行化。最后，如果流中的元素数量不能被 chunkSize 整除(在我的示例中为 4)，必须从 TaskMapper 中检索剩余的 future 。并在流外处理。

此构造适用于我执行的测试，但我知道由于流的副作用、状态完整和未定义的评估行为，它可能很脆弱。

编辑 2:

我使用自定义 Spliterator 制作了之前编辑的构造版本:

public class ExecutorServiceSpliterator<T> extends AbstractSpliterator<Future<T>>{
    private final Spliterator<? extends Callable<T>> srcSpliterator;
    private final ExecutorService service;
    private final int chunkSize;
    private final Queue<Future<T>> futures = new LinkedList<>();

    private ExecutorServiceSpliterator(Spliterator<? extends Callable<T>> srcSpliterator) {
        this(srcSpliterator, Executors.newFixedThreadPool(8), 30); //default
    }

    private ExecutorServiceSpliterator(Spliterator<? extends Callable<T>> srcSpliterator, ExecutorService service, int chunkSize) {
        super(Long.MAX_VALUE, srcSpliterator.characteristics() & ~SIZED & ~CONCURRENT);
        this.srcSpliterator = srcSpliterator;
        this.service = service;
        this.chunkSize = chunkSize;
    }

    public static <T> Stream<T> pipeParallelized(Stream<? extends Callable<T>> srcStream){
        return getStream(new ExecutorServiceSpliterator<>(srcStream.spliterator()));
    }

    public static <T> Stream<T> pipeParallelized(Stream<? extends Callable<T>> srcStream, ExecutorService service, int chunkSize){
        return getStream(new ExecutorServiceSpliterator<>(srcStream.spliterator(), service, chunkSize));
    }

    private static <T> Stream<T> getStream(ExecutorServiceSpliterator<T> serviceSpliterator){
        return StreamSupport.stream(serviceSpliterator, false)
            .map(future -> {
                try {               
                    return future.get();
                } catch (InterruptedException | ExecutionException ex) {    
                    throw new RuntimeException(ex);
                }
            }
        );
    }

    @Override
    public boolean tryAdvance(Consumer<? super Future<T>> action) {
        boolean didAdvance = true;
        while((didAdvance = srcSpliterator.tryAdvance(c -> futures.add(service.submit(c))))
                && futures.size() < chunkSize);
        if(!didAdvance){
            service.shutdown();
        }

        if(!futures.isEmpty()){
            Future<T> future = futures.remove();
            action.accept(future);
            return true; 
        }
        return false;
    }           

}

此类提供函数 ( pipeParallelized() )，它接收 Callable 的流-elements 并行执行它们，然后输出包含结果的顺序流。 Spliterators允许是有状态的。因此，希望这个版本不会违反任何流操作约束。 Splitterator是这样的可以使用(接近于“神奇的单线”):

ExecutorServiceSpliterator.pipeParallelized(stream);

此行采用 Callables 的流stream并行执行它们并返回一个包含结果的顺序流(管道延迟发生 -> 应该与数百万个可调用对象一起工作)可以通过常规流操作进一步处理。

执行ExecutorServiceSpliterator非常基本。它应该主要展示原则上如何做到这一点。可以优化服务的重新提供和结果的检索。例如，如果允许结果流无序，则 CompletionService可以使用。