使用带有 CompletableFuture 的默认通用 fork/join 池进行长时间阻塞调用是不好的做法吗？

Posted 2023-02-16

【中文标题】使用带有 CompletableFuture 的默认通用 fork/join 池进行长时间阻塞调用是不好的做法吗？

【英文标题】：Is it bad practice to use default common fork/join pool with CompletableFuture for doing long blocking calls?

【发布时间】：2018-01-28 08:14:00

【问题描述】：

假设我有一个 CompletableFuture，它包装了一个阻塞调用，比如使用 JDBC 查询后端。在这种情况下，由于我没有将任何执行程序服务作为参数传递给 CompletableFuture.supplyAsync()，因此通过后端获取资源的实际阻塞工作应该由公共 Fork/Join 池中的线程完成。 bad practice 不是让来自普通 FJpool 的线程进行阻塞调用吗？ 我的优势是我的主线程没有阻塞，因为我委托阻塞调用异步运行.检查 abt JDBC 调用是否阻塞 here 。如果这个推论是真的，为什么可以选择使用默认的通用 FJpool 和 CompletableFuture？

CompletableFuture<List<String>> fetchUnicorns  = 
    CompletableFuture.supplyAsync(() -> 
        return unicornService.getUnicorns();
    );

fetchUnicorns.thenAccept(/**Do something with the result*/);

【问题讨论】：

CompletableFuture 作为 API 有许多设计缺陷，这就是其中之一。是的，在任何线程池（包括 FJ 池）中混合阻塞和非阻塞任务是一种不好的做法。

【参考方案1】：

您不应使用阻塞调用（以这种方式）的原因是，公共池并行性已配置为利用现有 CPU 内核，假设非阻塞作业。阻塞的线程会降低使用同一池的其他任务的并行度。

但是有官方的解决方法：

class BlockingGetUnicorns implements ForkJoinPool.ManagedBlocker 
    List<String> unicorns;
    public boolean block() 
        unicorns = unicornService.getUnicorns();
        return true;
    
    public boolean isReleasable()  return false; 

CompletableFuture<List<String>> fetchUnicorns  = 
    CompletableFuture.supplyAsync(() -> 
        BlockingGetUnicorns getThem = new BlockingGetUnicorns();
        try 
            ForkJoinPool.managedBlock(getThem);
         catch (InterruptedException ex) 
            throw new AssertionError();
        
        return getThem.unicorns;
    );

ForkJoinPool.ManagedBlocker 是潜在阻塞操作的抽象，允许 Fork/Join 池在识别到工作线程即将被阻塞时创建补偿线程。

应该很明显，使用起来要容易得多

CompletableFuture<List<String>> fetchUnicorns  = 
    CompletableFuture.supplyAsync(() -> unicornService.getUnicorns(),
                                  Executors.newSingleThreadExecutor());

这里。在生产环境中，您将保留对执行程序的引用，重用它并最终在其上调用shutDown。对于执行器不被重用的用例，

CompletableFuture<List<String>> fetchUnicorns  = 
    CompletableFuture.supplyAsync(() -> unicornService.getUnicorns(),
                                  r -> new Thread(r).start());

这样就足够了，线程将在作业完成后自动释放。

【讨论】：

感谢您提供有关 ForkJoinPool.ManagedBlocker 的信息。有帮助！

这个答案真的很有帮助！谢谢。

【参考方案2】：

如果这个推论是真的，为什么可以选择使用默认的通用 FJpool 和 CompletableFuture？

因为并非所有工作都在阻塞。

您可以选择使用CompletableFuture.supplyAsync(Supplier<U>, Executor) 在自定义执行程序上安排您的阻止工作