我需要关闭在java中一致处理数据的ExecutorService吗?

Question

我有一个每24小时在不同线程中运行任务的Spirngboot应用程序。任务需要花费一点时间，但在那之后，在第二天的下一批处理触发api之前，应用程序就会变得空闲。演示代码如下:

private ExecutorService es = Executors.newFixedThreadPool(2);
@PostMapping
public void startTest(@RequestBody DummyModel dummy) throws InterruptedException {
int i = 0;
while(i<3) {
es.execute(new ListProcessing(dummy));
es.execute(new AnotherListProcessing(dummy));
i++;
}
// because methods are async this line is reached in an instant  before processing is done
}

现在你可以看到，在我的while循环之后没有es.shutdown()。在大多数文章和讨论中，似乎都强调在完成工作之后.shutdown()命令的重要性。在我的while循环之后添加它意味着我所做的下一个post请求将导致错误(这是有意义的，因为.shutdown()声明在现有任务完成后它将不允许新的任务)。

现在，我想知道在这里做.shutdown()真的很重要吗?我的应用程序每天都会收到一次post请求，因此ExecutorService将经常使用。在很长一段时间内不关闭你的执行人有什么缺点吗?如果我真的需要在每次使用ExecutorService时关闭它，我怎么做才能使应用程序在第二天准备好接收新的请求?

我正在考虑在我的while循环之后添加这些行:

es.shutdown();
es = Executors.newFixedThreadPool(2);

它工作，但似乎:

a)不必要的(为什么要关闭它并浪费精力重新创建它)

b)，只是看起来&感觉错了。一定有更好的办法。

所以似乎你可以创建一个自定义的ThreadPoolExecutor(这对于我简单的用例似乎有点小题大做)，或者你可以使用ExecutorService的CachedThreadPool选项(尽管它会尝试使用尽可能多的线程，所以如果你只需要使用n个线程，这个选项可能不适合你)。

更新2正如Thomas和Deinum所解释的，我们可以使用自定义的executor。它做的正是ExecutorService做的+清理，也允许快速&简单的配置方法。对于那些好奇的人，下面是我如何实现它的:

@Autowired
private TaskExecutor taskExecutor;
@PostMapping
public void startTest(@RequestBody DummyModel dummy) throws InterruptedException {
taskExecutor.execute(new ProcessList());
taskExecutor.execute(new AnotherProcessList());
taskExecutor.execute(new YetAnotherProcessList());
}

其中taskExecutor是在我的主类中定义的bean(或者它可以在任何带有@Configuration注释的类中定义)。内容如下:

@Bean
public TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5); // min number of threads that are always there
executor.setMaxPoolSize(10); // if threads are full and queue is full then additional threads will be created
executor.setQueueCapacity(5); // the number of tasks to be placed in the queue (caution queues require memory. Larger queue = more memory)
return executor;
}

Answer 1

你提出的解决方案不可靠。

es.shutdown();
es = Executors.newFixedThreadPool(2);

这假设方法startTest只会被一个传入请求并发调用，并且下一个传入的请求总是在执行器关闭并刷新之后。

只有在方法的范围内创建ExecutorService时，该解决方案才有效。然而，这也是有问题的。如果有100个请求，您将创建200个并发线程，每个线程占用资源。因此，您有效地创建了一个潜在的资源泄漏(或者至少为您的应用程序创建了一个攻击向量)。

一般的经验法则，如果你自己在同一作用域中创建了Executor，那么你应该关闭它，如果不是保持不变的话。在您的示例中，您基本上使用共享线程池，并且应该只在应用程序停止时执行关闭操作。你可以在控制器的@PreDestroy方法中实现

@PreDestroy
public void cleanUp() {
es.shutdown();
}

然而，除了将它添加到控制器中，你还可以将ExecutorService定义为bean并配置一个destroy方法。

@Bean(destroyMethod="shutdown")
public ExecutorService taskExecutor() {
return Executors.newFixedThreadPool(2);
}

你现在可以在你的控制器中依赖注入ExecutorService。

@RestController
public class YourController {
private final ExecutorService es;
public YourController(ExecutorService es) {
this.es=es;
}
}

最后，我怀疑您使用Spring (Boot)提供的TaskExecutor会更好，它会自动进入Spring上下文生命周期。你可以简单地把它注入你的控制器，而不是ExecutorService。

@RestController
public class YourController {
private final TaskExecutor executor;
public YourController(TaskExecutor executor) {
this.executor = executor;
}
}

Spring Boot默认提供了一个将被注入的，您可以通过使用spring.task.execution.pool.*属性来控制这些。

spring.task.execution.pool.max-size=10
spring.task.execution.pool.core-size=5
spring.task.execution.pool.queue-capacity=15

或者您可以定义一个bean，这也将覆盖默认的TaskExecutor。

@Bean
public ThreadPoolTaskExecutor taskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(5);
return executor;
}