java 多线程最佳实践：约束任务 newFixedThreadPool

Multithreading best practices : constraining tasks newFixedThreadPool

提问by FireFox

I want to launch a lot of tasks to run on a database of +-42Mio records. I want to run this in batches of 5000 records/time (results in 850 tasks). I also want to limit the number of threads (to 16) java starts to do this for me and I am using the current code to accomplish this task:

我想启动很多任务以在 +-42Mio 记录的数据库上运行。我想分批运行 5000 条记录/时间（导致 850 个任务）。我还想限制线程数（到 16 个）java 开始为我执行此操作，并且我正在使用当前代码来完成此任务：

 ExecutorService executorService = Executors.newFixedThreadPool(16);
 for (int j = 1; j < 900 + 1; j++) {
     int start = (j - 1) * 5000;
     int stop = (j) * 5000- 1;
     FetcherRunner runner = new FetcherRunner(routes, start, stop);
     executorService.submit(runner);

     Thread t = new Thread(runner);
     threadsList.add(t);
     t.start();
 }

Is this the correct way to do this? Particularly as I have the impression that java just fires away all tasks ...(FetcherRunnerimplements runnable)

这是正确的方法吗？特别是因为我的印象是 java 只会触发所有任务......（FetcherRunner实现runnable）

回答by Biju Kunjummen

The first part using ExecutorService looks good:

使用 ExecutorService 的第一部分看起来不错：

...
FetcherRunner runner = new FetcherRunner(routes, start, stop);
executorService.submit(runner);

The part with Thread should not be there, I am assuming you have it there just to show how you had it before?

Thread 的部分不应该在那里，我假设您在那里拥有它只是为了展示您以前是如何拥有它的？

Update:Yes, you don't require the code after executorService.submit(runner), that is going to end up spawning a huge number of threads. If your objective is to wait for all submitted tasks to complete after the loop, then you can get a reference to Futurewhen submitting tasks and wait on the Future, something like this:

更新：是的，您不需要之后的代码executorService.submit(runner)，这最终会产生大量线程。如果您的目标是在循环后等待所有提交的任务完成，那么您可以Future在提交任务时获取引用并等待Future，如下所示：

ExecutorService executorService = Executors.newFixedThreadPool(16);
List<Future<Result>> futures = ..;
 for (int j = 1; j < 900+ 1; j++) {
 int start = (j - 1) * 5000;
 int stop = (j) * 5000- 1;
 FetcherRunner runner = new FetcherRunner(routes, start, stop);
 futures.add(executorService.submit(runner));

}
for (Future<Result> future:futures){
    future.get(); //Do something with the results..
}

回答by Stephen C

Is this the correct way of working?

这是正确的工作方式吗？

The first part is correct. But you shouldn't be creating and starting new Thread objects. When you submit the Runnable, the ExecutorService puts it on its queue, and then runs it when a worker thread becomes available.

第一部分是正确的。但是您不应该创建和启动新的 Thread 对象。当您提交 Runnable 时，ExecutorService 将其放入其队列，然后在工作线程可用时运行它。

.... I use the threadlist to detect when all my threads are finished so I can continue processing results.

.... 我使用线程列表来检测我的所有线程何时完成，以便我可以继续处理结果。

Well if you do what you are currently doing, you are running each task twice. Worse still, the swarm of manually created threads will all try to run in parallel.

好吧，如果您执行当前正在执行的操作，那么您将每个任务运行两次。更糟糕的是，大量手动创建的线程都将尝试并行运行。

A simple way to make sure that all of the tasks have completed is to call awaitTermination(...)on the ExecutorService. (An orderly shutdown of the executor service will have the same effect ... if you don't intend to use it again.)

确保所有任务都已完成的一种简单方法是调用awaitTermination(...)ExecutorService。（有序关闭 executor 服务将具有相同的效果......如果您不打算再次使用它。）

The other approach is to create a Futurefor each FetcherRunner's results, and attempt to getthe result after all of the tasks have been submitted. That has the advantage that you can start processing early results before later ones have been produced. (However, if you don't need to ... or can't ... do that, using Futures won't achieve anything.)

另一种方法是Future为每个FetcherRunner的结果创建一个，并get在所有任务提交后尝试结果。这样做的好处是您可以在生成较晚的结果之前开始处理早期结果。（但是，如果您不需要……或不能……这样做，则使用 Futures 将无法实现任何目标。）

回答by chubbsondubs

You don't need to the part after the call to submit. The code you have that creates a Thread will result in 900 threads being created! Yowza. The ExecutorService has a pool of 16 threads and you can run 16 jobs at once. Any jobs submitted when all 16 threads are busy will be queued. From the docs:

您不需要在调用后提交的部分。您拥有的创建线程的代码将导致创建 900 个线程！约扎。ExecutorService 有一个包含 16 个线程的池，您可以一次运行 16 个作业。当所有 16 个线程都忙时提交的任何作业都将排队。从文档：

Creates a thread pool that reuses a fixed number of threads operating off a shared unbounded queue. At any point, at most nThreads threads will be active processing tasks. If additional tasks are submitted when all threads are active, they will wait in the queue until a thread is available. If any thread terminates due to a failure during execution prior to shutdown, a new one will take its place if needed to execute subsequent tasks. The threads in the pool will exist until it is explicitly shutdown.

创建一个线程池，该线程池重用固定数量的线程在共享的无界队列中运行。在任何时候，最多有 nThreads 个线程是活动的处理任务。如果在所有线程都处于活动状态时提交了额外的任务，它们将在队列中等待，直到有线程可用。如果任何线程在关闭前的执行过程中由于失败而终止，则在需要执行后续任务时，将有一个新线程代替它。池中的线程将一直存在，直到它被明确关闭。

So there is no need for yet another thread. If you need to be notified after a task has finished you can have it call out. Other options are to cache all of the Future's returned from submit, and upon each task being finished you can check to see if all Future's are done. After all Future's are finished you can dispatch another function to run. But it will run ON one of the threads in the ExecutorService.

所以不需要另一个线程。如果您需要在任务完成后收到通知，您可以调用它。其他选项是缓存从提交返回的所有 Future，并且在每个任务完成后，您可以检查是否所有 Future 都已完成。在所有 Future 完成后，您可以调度另一个函数来运行。但它将在 ExecutorService 中的线程之一上运行。

回答by Amareswar

The best way would be to use countdownlatch as follows

最好的方法是使用 countdownlatch 如下

    ExecutorService executorService = Executors.newFixedThreadPool(16);
  CountdownLatch latch = new CountdownLatch(900);
 FetcherRunner runner = new FetcherRunner(routes, start, stop, latch);
 latch.await();

in the FetcherRunner under finally block use latch.countDown();code after await()will be executed only when all the tasks are completed.

在 FetcherRunner 下的 finally 块中使用的latch.countDown();代码await()只有在所有任务完成后才会执行。

回答by Jason

Changed from your code:

从您的代码更改：

    ExecutorService executorService = Executors.newFixedThreadPool(16);
    for (int j = 1; j < 900 + 1; j++) {
        int start = (j - 1) * 5000;
        int stop = (j) * 5000 - 1;
        FetcherRunner runner = new FetcherRunner(routes, start, stop);
        executorService.submit(runner);

    }