哪个执行人员服务最适合阻止IO任务

发布于 2025-01-26 13:11:03 字数 913 浏览 3 评论 0原文

让我们想象，我们有n个独立的阻止IO任务，例如将rest通用到另一台服务器的任务。然后所有答案我们都需要组合。每个任务都可以处理10秒以上。

我们可以顺序处理它，并在末尾花费〜n*10秒：

  task1ans task1 = service1.dosomething（）;
task2ans task2 = service2.dosomething（）
...
返回结果；

另一个策略是使用完整的future以并行方式处理它，并在所有任务上花费了〜10秒：

 完整的future＆lt; task1ans＆gt; task1cs = ploctableFuture.supplyAsync（（（） - ＆gt; service1.dosomething（），bestExeCutor）;
完整的future＆lt; task2ans＆gt; task2cs = ploteableFuture.supplyAsync（（） - ＆gt; service2.dosomething（），bestExeCutor）;
返回完整的future.allof（task1cs，task2cs）
   。
       ...
       //将任务1，任务2组合到结果对象中
       返回结果；
   }）。加入（）;

第二种方法具有好处，但我不明白哪种类型的线程池是适合此类任务的最好的：

ExecutorService bestExecutor = Executors.newFixedThreadPool(30)   /// or Executors.newCachedThreadPool() or Executors.newWorkStealingPool()

我的问题是哪种方法是执行人员服务最适合过程N-并行阻止IO任务。

原文

Let's imagine that we have n independent blocking IO tasks e.g. tasks for rest-call to another server. Then all answer we need to combine. Every task can be processing over 10 second.

We can process it sequentially and spent ~n*10 second at the end:

Task1Ans task1 = service1.doSomething();
Task2Ans task2 = service2.doSomething()
...
return result;

Another strategy is to process it in parallel manner using CompletableFuture and spent ~ 10 second on all task:

CompletableFuture<Task1Ans> task1Cs = CompletableFuture.supplyAsync(() -> service1.doSomething(), bestExecutor);
CompletableFuture<Task2Ans> task2Cs = CompletableFuture.supplyAsync(() -> service2.doSomething(), bestExecutor);
return CompletableFuture.allOf(task1Cs, task2Cs)
   .thenApply(nothing -> {
       ...
       // combine task1, task2 into result object
       return result;
   }).join();

The second approach has benefits, but I can't understand which type of thread pool is the best for this kind of task:

ExecutorService bestExecutor = Executors.newFixedThreadPool(30)   /// or Executors.newCachedThreadPool() or Executors.newWorkStealingPool()

My question is which ExecutorService is best for process n-parallel blocking IO tasks.

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

瘫痪情歌 2025-02-02 13:11:03

在CPU绑定的任务上，您不会通过比CPU内核更多的线程获得其他性能。因此，在这种情况下，8 Core / 8线程CPU仅需8个线程即可最大程度地提高性能，并通过更多信息来失去性能。与CPU内核相比，IO任务通常可以通过使用大量线程来获得性能，因为CPU时间可以在等待IO时进行其他事情。但是，即使每个线程的CPU开销较低，每个线程都会进食内存，并且会限制缩放，并会产生缓存/上下文开关。

鉴于您的任务是io限制的，并且您没有提供任何其他约束，您应该应该可能只是为您的每个IO任务运行不同的线程。您可以使用固定或缓存的线程池来实现这一目标。

如果您的IO任务的数量很大（千+），则应限制线程池的最大尺寸，因为您可以拥有太多的线程。

如果您的任务是CPU绑定的，则应再次将线程池限制为更小的尺寸。可以通过以下方式动态获取

int cores = Runtime.getRuntime().availableProcessors();

可以通过使用：同样，就像您的CPU具有缩放限制一样，您的IO设备通常也具有缩放限制，也。您不应该超过该限制，但是如果不衡量，很难说限制在哪里。

On completely CPU bound tasks you do not get additional performances by going with more threads than CPU cores. So in this scenario, 8 core / 8 thread CPU needs only 8 thread to maximize performances, and loses performance by going with more. IO tasks usually do gain performances by going with larger number of threads than CPU cores, as CPU time is available to do other stuff while waiting for IO. But even when CPU overhead of each thread is low there are limits to scaling as each thread eats into memory, and incurs caching/context switches..

Given that your task is IO limited, and you didn't provide any other constraints, you should probably just run different thread for each of your IO tasks. You can achieve this by either using fixed or cached thread pool.

If the number of your IO tasks is very large (thousands+), you should limit the maximum size of your thread pool, as you can have such thing as too many of threads.

If your task are CPU bound, you should again limit thread pool to even smaller size. Number of cores can be dynamically fetched by using:

int cores = Runtime.getRuntime().availableProcessors();

Also, just as your CPU has scaling limit, your IO device usually has a scaling limit too. You should not exceed that limit, but without measuring it is hard to say where limit is.

回复收藏 0 原文

月隐月明月朦胧 2025-02-02 13:11:03

Project Loom

您的情况适合使用为Java的未来版本提出的新功能： and 结构化的并发。这些是 project loom 。

当今的Java线程被映射到主机操作系统线程上。当Java代码块时，主机线程块。主机操作系统坐着空闲，等待执行恢复。主机操作系统线程是重量级的，就CPU和内存而言。因此，这个空转不是最佳的。

相比之下，项目织机中的虚拟线程被映射到主机OS线程上。当在虚拟线程块中代码时，该任务是“停放”的，请放置以允许另一个虚拟线程的任务在某些执行时间。虚拟线程的这种停车位在JVM内进行管理，因此在CPU和内存中，它都非常优化，非常快，非常有效。结果，在通用硬件上运行的Java应用程序一次可以支持数千个虚拟线程。

executorService是 Autoclosable 在Loom中。因此，我们可以使用try-with-resources在try（executorService es = executors.newvirtualThreadPertasKexecutor（））中包含您的整个任务。完成后，控制流程从try-with-resources块中退出，您知道自己的任务已完成。访问未来对您提交的每个任务返回的对象。无需完整的future。

现在在Java 19和20中预览和孵育织机功能。虚拟线程功能是计划在Java 21 <21 < /a>。

有关更多信息，请参阅与项目LOOM团队成员的几篇文章，演讲和访谈。其中包括罗恩·波特勒（Ron Pressler）和艾伦·贝特曼（Alan Bateman）。并查看jepCafé的相关插曲。

回复收藏 0 原文

绝對不後悔。 2025-02-02 13:11:03

如果我正确理解您的问题，则无论选择executorService，都更重要的是如何调用executorService。

例如

ExecutorService executorService=Executors.newCachedThreadPool();
executorService.invokeAll(..);

，现在在这里，InvokeAll（..）将阻止直到完成内部提供的所有任务。
因此，我觉得选择任何执行人员服务＆amp;调用InvokeAll（..）将适合您的要求。

另外，请查看此 se问题讨论新Java 8 ＆amp; InvoKeall。

If I understand your question properly, for above behaviour, irrespective of selection of executorService, it is more important how you are calling your executorService.

E.g.

ExecutorService executorService=Executors.newCachedThreadPool();
executorService.invokeAll(..);

Now here, invokeAll(..) will block until all supplied tasks inside are completed.
So I feel selecting any ExecutorService & calling invokeAll(..) will be suitable for your requirement.

Also please have a look at this SE Question which discusses new Java 8 introduction of ExecutorCompletionService & invokeAll.

回复收藏 0 原文

泼猴你往哪里跑 2025-02-02 13:11:03

我找到了此类任务的最佳解决方案。要找到解决方案的目的是查看执行者的实现。newcachedThreadPool（）或opecutors.newfixedthreadpool（30）

   public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

我的决定是直接实例化ThreadPoolPoolExecutor，并设置可以由线程池创建的线程的上限。并在未使用的线程后设置超时

int nThread = 90;
long timeoutSec = 120;
ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setNameFormat("Executor-Worker-%d")
                .setDaemon(true)
                .build();
Executor delegate = new ThreadPoolExecutor(
    0,  // min number of thread in pool
    nThread, // max number of thread in pool
    timeoutSec, // terminate idle thread after
    TimeUnit.SECONDS,
    new SynchronousQueue<Runnable>(),
    threadFactory
);

I found the optimal solution for this kind of task. All I nead to find the solution is to look at implementation of Executors.newCachedThreadPool() or Executors.newFixedThreadPool(30)

   public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

My decision is to instantiate ThreadPoolExecutor directly and set upper bound of threads that can be created by thread pool. And set timeout after unused threads can be terminated

int nThread = 90;
long timeoutSec = 120;
ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setNameFormat("Executor-Worker-%d")
                .setDaemon(true)
                .build();
Executor delegate = new ThreadPoolExecutor(
    0,  // min number of thread in pool
    nThread, // max number of thread in pool
    timeoutSec, // terminate idle thread after
    TimeUnit.SECONDS,
    new SynchronousQueue<Runnable>(),
    threadFactory
);

回复收藏 0 原文

~没有更多了~