6

假设我有一个并行度为 n 的 ForkJoinPool 设置，并且我将这样的并行计算称为：

 workpool.submit(
            () -> {
                    objects.values().parallelStream().forEach(obj -> {
                        obj.foo();
                    });
                });

我这样做是为了确保在工作池中创建在那里生成的线程（我有需要隔离的系统的不同组件）。现在假设调用 this 的线程也在这个工作池中执行，我这样做：

 Future<?> wait =  workpool.submit(
            () -> {
                    objects.values().parallelStream().forEach(obj -> {
                        obj.foo();
                    });
                });
 wait.get()

1) 我是否阻塞了 ForkJoinPool 中的线程？如果我让 n 个线程全部阻塞在 future 上，同时尝试在工作池中安排任务，这会导致死锁吗？我不清楚 ForkJoinPool 中的“最大并行度”是否意味着（如果有 n 个非阻塞任务），将始终有 n 个线程在执行，或者是否有固定数量的线程，无论是否有被封锁。如果我改用 wait.join() 会怎样wait.join（我不需要检查异常，因为此代码中抛出的任何异常都已经生成运行时异常。如果我理解正确，join() 将允许线程在等待时执行排队的任务）

2) 如果我通过 () -> {} 创建一个可运行的“包装器”类，我是否还能从并行流的轻量级 forkjoin 任务中受益？

3）使用它是否有任何缺点/优点（假设 .join() 确实实现了我认为的工作窃取行为）：

        CompletableFuture.supplyAsync(this::mylambdafunction, workpool)  
             .thenAccept(this::mynextfunction);
4

1 回答 1

1

对第 1 点的回应：如果没有看到实际的方法实现，很难知道您的代码是否会阻塞。处理阻塞代码的一种方法是增加 forkjoin 线程池中的线程数。通常，对于计算密集型任务，forkjoin 线程中的线程数为 n+1，其中 n=处理器数。或者，如果您有 I/O 阻塞，您可以使用 ManagedBlocker。

对第 2 点的回应：是

对第 3 点的回应：您的 completableFuture 代码的明显优势是 thenAccept 是非阻塞的。因此，控制将立即通过您的 CompletableFuture 块进入下一个语句，而无需等待，而在您使用 ForkJoin 池编写的早期代码中，wait.get() 将阻塞，直到您获得答案并且直到那时才会继续。

于 2017-04-08T20:08:20.730 回答