Spring Boot REST - 请求未使用 ThreadPoolTask​​Executor 配置执行

Question

我正在尝试开发一个 Spring Boot 应用程序。我在没有spring框架的情况下用核心java编写了所有核心实现。我在这个 Spring Boot 应用程序中使用了那个 jar。我想管理我的休息控制器的并发性。因此，在主类中相应地配置了 ThreadPoolTask​​Executor。理想情况下，我只希望 2 个并发请求进入 execute() 方法，我注释了 Async。我一次测试 2 个并发请求，但我在日志中看到我的请求同时输入 execute()。所有任务都是内存密集型的。所以这些都因堆内存问题而失败。我试图找出理想的并发数。我想知道我的配置是正确的还是我遗漏了什么？谢谢。

这是我的主要课程：

@SpringBootApplication
@EnableAsync
public class RestapiApplication implements AsyncConfigurer {

    public static void main(String[] args) {
        ApplicationContext ctx = SpringApplication.run(RestapiApplication.class, args);
        System.out.println("Rightdata Middleware ready to accept requests:");
    }

    @Bean(name = "executor1")
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setMaxPoolSize(2);
        taskExecutor.setCorePoolSize(2);
        taskExecutor.setThreadNamePrefix("LULExecutor-");
        taskExecutor.setQueueCapacity(100);
        taskExecutor.initialize();
        return taskExecutor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new SimpleAsyncUncaughtExceptionHandler();
    }
}

这是我的 REST 控制器：

@RestController
@RequestMapping("/end2end")
public class End2EndRestController {

    /**
     * The log.
     */
    private final Logger log = LoggerFactory.getLogger(this.getClass());

    @RequestMapping(method = RequestMethod.POST)
    public JSONObjectPOJO process(@RequestBody String end2EndScenarioString) throws InterruptedException, ExecutionException {

        final JSONObjectPOJO jsonObjectPOJO = convertToJavaObject(end2EndScenarioString);
        final ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    execute(jsonObjectPOJO);
                } catch (Exception e) {
                    e.getMessage();
                }
            }});
            executor.shutdown();
            return jsonObjectPOJO;
        }

    @Async("executor1")
    private void execute(JSONObjectPOJO jsonObjectPOJO) throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Future<?> futureTarget;
        Future<?> futureSource;
        futureSource = processSource(executorService);
        futureTarget = processTarget(executorService);
        manageSourceProcessingResults(futureSource);
        manageTargetProcessingResults(futureTarget);
        executorService.shutdown();
        //Do rest of the tasks.
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    protected Future<?> processSource(executorService){
        //Get appropriate class instance with call() - coreActionClass.
        Future<?> futureSource = executorService.submit(coreActionClass);
        return futureSource;
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    protected Future<?> processTarget(executorService){
        //Get appropriate class instance with call() - coreActionClass.
        Future<?> futureTarget = executorService.submit(coreActionClass); //callable method in core.
        return futureTarget;
    }

    private void manageSourceProcessingResults(Future<?> futureSource) {
        try{
            futureSource.get();
        } catch(Exception e){
            e.printStackTrace();
        }
    }

    private void manageTargetProcessingResults(Future<?> futureTarget) {
        try{
            futureTarget.get();
        } catch(Exception e){
            e.printStackTrace();
        }
    }
}

更新- 1：

我现在已将代码更改为以下内容：

@RestController
@RequestMapping("/end2end")
public class End2EndRestController {

    /**
     * The log.
     */
    private final Logger log = LoggerFactory.getLogger(this.getClass());

    @RequestMapping(method = RequestMethod.POST)
    public JSONObjectPOJO process(@RequestBody String end2EndScenarioString) throws InterruptedException, ExecutionException {

        final JSONObjectPOJO jsonObjectPOJO = convertToJavaObject(end2EndScenarioString);
        final ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    execute(jsonObjectPOJO);
                } catch (Exception e) {
                    e.getMessage();
                }
            }});
            executor.shutdown();
            return jsonObjectPOJO;
        }    
}

和 AsyncService 类：

public class AsyncService {

    @Async("executor1")
    public void execute(JSONObjectPOJO jsonObjectPOJO) throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Future<?> futureTarget;
        Future<?> futureSource;
        futureSource = processSource(executorService);
        futureTarget = processTarget(executorService);
        manageSourceProcessingResults(futureSource);
        manageTargetProcessingResults(futureTarget);
        executorService.shutdown();
        //Do rest of the tasks.
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    protected Future<?> processSource(executorService){
        //Get appropriate class instance with call() - coreActionClass.
        Future<?> futureSource = executorService.submit(coreActionClass);
        return futureSource;
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    protected Future<?> processTarget(executorService){
        //Get appropriate class instance with call() - coreActionClass.
        Future<?> futureTarget = executorService.submit(coreActionClass); //callable method in core.
        return futureTarget;
    }

    private void manageSourceProcessingResults(Future<?> futureSource) {
        try{
            futureSource.get();
        } catch(Exception e){
            e.printStackTrace();
        }
    }

    private void manageTargetProcessingResults(Future<?> futureTarget) {
        try{
            futureTarget.get();
        } catch(Exception e){
            e.printStackTrace();
        }
    }
}

1. 我的理解是当我配置maxpoolsize(2)时不再 一次在 execute() 方法中将有超过 2 个请求。为一个 新请求进入，较早的请求之一必须完成 它的执行。我的理解正确吗？ async 是否适用 到内部执行器服务？
2. 我认为一次只处理 2 个请求，并且 这些请求中的每一个都可以产生 2 个不同的线程并完成 它的任务。请澄清。

Answer 1

我看到了两个问题。

1) 在您的 process 方法中，您正在创建一个新的 ExecutorService。这不是必需的。相反，只需在检索到 jsonObjectPOJO 后调用 execute 方法即可。

2) 你不能在实现它的同一个类中使用@Async。您需要创建一个新类，让我们调用MyAsyncService 来包含@Async 方法。这是因为面向方面的编程在幕后进行。

查看this link 了解更多信息。以下是链接中的引用。

首先——让我们回顾一下规则——@Async 有两个限制：

它只能应用于公共方法 自调用——从同一个类中调用异步方法——不起作用 原因很简单——方法需要公开才能被代理。而自调用也不起作用，因为它绕过了代理，直接调用了底层方法。

编辑 1：

@RestController
@RequestMapping("/end2end")
public class End2EndRestController {

@AutoWired
AsyncService asyncService;

private final Logger log = LoggerFactory.getLogger(this.getClass());
@RequestMapping(method = RequestMethod.POST)
public JSONObjectPOJO process(@RequestBody String end2EndScenarioString) throws InterruptedException, ExecutionException {
    final JSONObjectPOJO jsonObjectPOJO = convertToJavaObject(end2EndScenarioString);
    asyncService.execute(jsonObjectPOJO);
    return jsonObjectPOJO;
}




public class AsyncService {

    @Async("executor1")
    public void execute(JSONObjectPOJO jsonObjectPOJO) throws Exception {
        //No Futures, no ExecutorServices, just process that request.
    }

}

通过创建和配置 ThreadPoolTask​​Executor 以仅使用 2 个线程，您已经实现了目标。

EDIT2：Spring @Async limit number of threads