如何在 Spring Boot @Async 中使用 ForkJoinPool？

Question

我想在我的 Spring Boot 项目中使用 ForkJoinPool，并带有 @Async 注释，例如 ThreadPoolTask​​Executor

例如：-

 @Bean("threadPoolTaskExecutor")
    public TaskExecutor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(1000);
        executor.setThreadNamePrefix("Async-");
        return executor;
    }

我在我的代码中使用https://dzone.com/articles/spring-boot-creating-asynchronous-methods-using-as 这个链接，但我想像这样使用 ForkJoinPool。

Answer 1

可以将 ForkJoinPool 与异步一起使用。这种方法不完全是 ForkJoinPool 的设计目的——解决分而治之任务并使用工作窃取算法，而是执行事件样式或 IO 阻塞任务，这是一个有效的用例。

所以有了基本用法，你可以这样做：

CompletableFuture.runAsync(() -> doSomething(), ForkJoinPool.commonPool());

您还可以在异步模式下创建自定义 ForkJoinPool。异步模式下的 ForkJoinPool 中的工作人员按 FIFO（先进先出）的顺序处理任务。默认情况下，ForkJoinPools 按 LIFO（后进先出）顺序处理此类任务。此外，异步模式设置仅涉及从未加入的分叉任务。可用于已提交但从未加入的事件式任务（因其副作用而执行的任务，而不是返回将由分叉任务处理然后加入的结果）。

您可以像这样在异步模式下创建具有给定并行度的自定义 ForkJoinPool（第一个参数是池大小，最后一个 - bool asyncMode）：

ForkJoinPool pool = new ForkJoinPool(
             6, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true);

那么你所追求的是：

    @Bean("threadPoolTaskExecutor")
    public Executor getAsyncExecutor() {
    ForkJoinPool pool = new ForkJoinPool(
             6, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true);
        return pool;
    }

Answer 2

在我看来，您不能将@Async 注释与ForkJoinPool 类结合使用。 ForkJoinPool 专用于将一个任务拆分为子任务以并行方式执行并合并为结果。并行流在幕后是usingForkJoinPool。要使用ForkJoinPool 执行任务，您需要实现RecursiveTask 或RecursiveAction 接口。我不知道我们如何一起使用这些东西。

Answer 3

我不知道你为什么要这样做，但你仍然可以使用ForkJoinPool 代替@Async，因为ForkJoinPool 正在实现Executor 的类

但是如果我们查看ForkJoinPool-documentation，我看不到任何设置线程池大小的方法，使用它是你的风险。

配置

@Bean("threadPoolTaskExecutor")
public Executor getAsyncExecutor() {
    ForkJoinPool pool = new ForkJoinPool();
    return pool;
}

服务

@Async("threadPoolTaskExecutor")
public void testRun()  {
    System.out.println(Thread.currentThread().getName());
}

输出

ForkJoinPool-1-worker-1

如果您对使用 ForkJoinPool 非常感兴趣，请使用 CompletableFuture.async 方法而不是 spring boot @Async docs

所有没有显式 Executor 参数的异步方法都使用 ForkJoinPool.commonPool() 执行（除非它不支持至少两个并行级别，在这种情况下，会创建一个新线程来运行每个任务）

Answer 4

要启用 Spring 的异步方法执行能力，您可以在配置类中使用 @EnableAsync 注解。

@Configuration
@EnableAsync
public class ThreadConfig {
    @Bean
    public TaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(4);
        executor.setThreadNamePrefix("prate");
        executor.initialize();
        return executor;
    }
}

在某些情况下，您不希望同一个线程池运行应用程序的所有任务。您可能需要具有不同配置的单独线程池来支持我们的功能。 所以你可以配置特定的ThreadPoolExecutor like -

@Configuration
@EnableAsync
public class ThreadConfig {
    @Bean(name = "specificTaskExecutor")
    public TaskExecutor specificTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.initialize();
        return executor;
    }
}

现在该函数应该设置限定符值以确定特定执行器或任务执行器的目标执行器。

@Async("specificTaskExecutor")
public void runFromAnotherThreadPool() {
    System.out.println("test here");
}