【问题标题】：ExecutorService should wait until batch of taksk is finished before starting againExecutorService 应该等到一批任务完成后再重新开始
【发布时间】：2021-06-17 01:59:29
【问题描述】：

问题：

我正在解析一个大型日志文件（大约 625_000_000 行）并将其保存到数据库中。

public class LogScheduler {

static int fileNumber = 1;

public Importer(IRequestService service) {
    this.service = service;
}

@Override
public  void run() {
    try {
        service.saveAll(getRequestListFromFile("segment_directory/Log_segment_"+fileNumber+".txt"));
        
    } catch (IOException e) {
        e.printStackTrace();
    }

   }
}

运行这个线程的方法是：

 public void scheduledDataSave() throws InterruptedException {
    int availableCores = Runtime.getRuntime().availableProcessors();
    String directory = "segment_directory";

    int filesInDirectory = Objects.requireNonNull(new File(directory).list()).length;

    ExecutorService executorService = Executors.newFixedThreadPool(availableCores);

    for (int i = 1; i <= filesInDirectory; i++) {
        executorService.execute(new Importer(service));
    }
    executorService.shutdown();

    }

在每个线程执行后在executorService.execute(new Importer(service)); 休眠之后插入Thread.sleep(); 方法，而不是像应该的8 个线程，因为它们在Executorservice 中而且我不知道为什么会发生这种情况，因为它不应该那样做。据我了解，ExecutorService 应该并行运行 8 个线程，完成它们，休眠，然后再次启动池。

每8个线程后如何“睡觉”？

【问题讨论】：

The problem is the CPU spike - 为什么？
@Eugene 我想说恒定的 100% CPU。我想找到一种方法来减少它，并结合睡眠来这样做
这可能与ExecutorService 无关，而是在所有这些文件之后需要清理的垃圾收集线程
你的问题不清楚。与其谈论您如何修改显示的代码，不如向我们展示您实际运行的代码。告诉我们你的目标，你想要达到的目标，因为那是模糊和不清楚的。由于不清楚，我是第二个投票结束。如果已关闭，请参与编辑您的问题，因为如果缺陷得到修复，它可以重新打开。
@BasilBourque 感谢您的指点。希望这更清楚

标签： java multithreading memory-leaks

【解决方案1】：

休眠提交任务的线程不会休眠提交的任务

您的问题不清楚，但显然围绕您的期望，即在每次调用 executorService.execute 后添加 Thread.sleep 将使执行程序服务的所有线程休眠。

    for ( int i = 1 ; i <= filesInDirectory ; i++ ) {
        executorService.execute( new Importer( service ) );   // Executor service assigns this task to one of the background threads in its backing pool of threads.
        Thread.sleep( Duration.ofMillis( 100 ).toMillis() ) ; // Sleeping this thread doing the looping. *Not* sleeping the background threads managed by the executor service.
    }

你的期望不正确。

Thread.sleep 正在休眠执行for 循环的线程。

执行器服务有自己的后台线程池。这些线程不受Thread.sleep 的影响，是其他线程。只有当您在每个线程上运行的代码中调用Thread.sleep 时，这些后台线程才会休眠。

因此，您将第一个任务提供给 executor 服务。执行器服务立即将该工作分派到其支持线程之一。该任务会立即执行（如果线程立即可用，并且没有被之前的任务占用）。

分配该任务后，您的for 循环将休眠一百毫秒，在此处显示的示例代码中。当for 循环处于休眠状态时，没有进一步的任务被分配给执行器服务。但是，当for 循环处于休眠状态时，提交的任务正在后台线程上执行。那个后台线程没有休眠。

最终，您的for 循环线程唤醒，分配第二个任务，然后重新进入睡眠状态。同时后台线程全速向前执行。

所以睡眠提交任务的线程不会睡眠已经提交的任务。

等待提交的任务完成

你的标题要求：

ExecutorService 应该等到一批 taksk 完成后再重新开始

提交任务后，在您的执行器服务上调用shutdown 和awaitTermination。在这些调用之后，您的代码会阻塞，等待所有提交的任务完成/取消/失败。

ExecutorService executorService = Executors.newVirtualThreadExecutor() ;
… submit tasks to that executor service …
executorService.shutdown() ;
executorSerivce.awaitTermination() ;  // At this point, the flow-of-control blocks until the submitted tasks are done.
System.out.println( "INFO - Tasks on background threads are done. " + Instant.now() );

我建议使用ExecutorService#submit 方法而不是ExecutorService#execute 方法。不同之处在于第一个方法返回一个Future 对象。您可以在向执行器服务提交任务时收集这些Future 对象。在shutdown 和awaitTermination 之后，您可以检查Future 对象的集合以检查它们的完成状态。

织机项目

如果Project Loom 成功，这样的代码会更简单、更清晰。 Project Loom 技术的实验版本是 available now，基于早期访问 Java 17。Loom 团队现在寻求反馈。

使用 Project Loom，ExecutorService 变为 AutoCloseable。这意味着我们可以使用 try-with-resources 语法在 ExecutorService 上自动调用新的 close 方法。这个close 方法首先阻塞，直到所有任务完成/取消/失败，然后关闭执行器服务。无需致电shutdown 或awaitTermination。

顺便说一句，Project Loom 还带来了虚拟线程（纤维）。这可能会显着提高代码的性能，因为它涉及存储 i/o 和数据库访问的大量阻塞。

try (
        ExecutorService executorService = Executors.newVirtualThreadExecutor() ;
)
{
    … submit tasks to that executor service …
}
// At this point, with Project Loom technology, the flow-of-control blocks until the submitted tasks are done.
// Also, the `ExecutorService` is automatically closed/shutdown by this point, via try-with-resources syntax.
System.out.println( "INFO - Tasks on background threads are done. " + Instant.now() );

使用 Project Loom，您可以使用与上述相同的方式收集返回的 Future 对象以检查完成状态。

您的代码中还有其他问题。但是你没有透露足够的信息来解决所有问题。

【讨论】：

【解决方案2】：

每8个线程后如何“睡觉”？

因此，如果您正在做这样的事情，那么它并没有按照您的想法做。

for (int i = 1; i <= filesInDirectory; i++) {
    executorService.execute(new Importer(service));
    Thread.sleep(...);
}

这会导致正在启动后台作业的线程进入休眠状态，并且不会影响每个作业的运行。我相信您缺少的是等待线程池完成：

 executorService.shutdown();
 executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);

这会等待线程池中的所有作业完成后再继续。

还有一件事。我使用executorService.submit(...) 与execute(...)。这是description of their difference。对我来说，另一个区别是使用execute(...) 运行的任务引发的任何异常都会导致正在运行的线程终止并可能重新启动。使用submit(...)，它允许您在需要时获取该异常，并阻止线程不必要地重新生成。

如果您能详细说明您想要完成的工作，我们应该能够提供帮助。

【讨论】：