码迷，mamicode.com
首页 > 编程语言 > 详细

如何合理地估算线程池大小？

时间：2018-11-09 16:25:34      阅读：239      评论：0      收藏：0      [点我收藏+]

标签：单线程   line   imu   验证   不能   cond   target   试验   puts   

如何合理地估算线程池大小？

感谢网友【蒋小强】投稿。

如何合理地估算线程池大小？

这个问题虽然看起来很小，却并不那么容易回答。大家如果有更好的方法欢迎赐教，先来一个天真的估算方法：假设要求一个系统的TPS（Transaction Per Second或者Task Per Second）至少为20，然后假设每个Transaction由一个线程完成，继续假设平均每个线程处理一个Transaction的时间为4s。那么问题转化为：

如何设计线程池大小，使得可以在1s内处理完20个Transaction？

计算过程很简单，每个线程的处理能力为0.25TPS，那么要达到20TPS，显然需要20/0.25=80个线程。

很显然这个估算方法很天真，因为它没有考虑到CPU数目。一般服务器的CPU核数为16或者32，如果有80个线程，那么肯定会带来太多不必要的线程上下文切换开销。

再来第二种简单的但不知是否可行的方法（N为CPU总核数）：

  • 如果是CPU密集型应用，则线程池大小设置为N+1
  • 如果是IO密集型应用，则线程池大小设置为2N+1

如果一台服务器上只部署这一个应用并且只有这一个线程池，那么这种估算或许合理，具体还需自行测试验证。

接下来在这个文档：服务器性能IO优化 中发现一个估算公式：

1 最佳线程数目 = （（线程等待时间+线程CPU时间）/线程CPU时间 ）* CPU数目

比如平均每个线程CPU运行时间为0.5s，而线程等待时间（非CPU运行时间，比如IO）为1.5s，CPU核心数为8，那么根据上面这个公式估算得到：((0.5+1.5)/0.5)*8=32。这个公式进一步转化为：

1 最佳线程数目 = （线程等待时间与线程CPU时间之比 + 1）* CPU数目

可以得出一个结论：

线程等待时间所占比例越高，需要越多线程。线程CPU时间所占比例越高，需要越少线程。

上一种估算方法也和这个结论相合。

一个系统最快的部分是CPU，所以决定一个系统吞吐量上限的是CPU。增强CPU处理能力，可以提高系统吞吐量上限。但根据短板效应，真实的系统吞吐量并不能单纯根据CPU来计算。那要提高系统吞吐量，就需要从“系统短板”（比如网络延迟、IO）着手：

  • 尽量提高短板操作的并行化比率，比如多线程下载技术
  • 增强短板能力，比如用NIO替代IO

第一条可以联系到Amdahl定律，这条定律定义了串行系统并行化后的加速比计算公式：

1 加速比=优化前系统耗时 / 优化后系统耗时

加速比越大，表明系统并行化的优化效果越好。Addahl定律还给出了系统并行度、CPU数目和加速比的关系，加速比为Speedup，系统串行化比率（指串行执行代码所占比率）为F，CPU数目为N：

1 Speedup <= 1 / (F + (1-F)/N)

当N足够大时，串行化比率F越小，加速比Speedup越大。

写到这里，我突然冒出一个问题。

是否使用线程池就一定比使用单线程高效呢？

答案是否定的，比如Redis就是单线程的，但它却非常高效，基本操作都能达到十万量级/s。从线程这个角度来看，部分原因在于：

  • 多线程带来线程上下文切换开销，单线程就没有这种开销
  • 锁

当然“Redis很快”更本质的原因在于：Redis基本都是内存操作，这种情况下单线程可以很高效地利用CPU。而多线程适用场景一般是：存在相当比例的IO和网络操作。

所以即使有上面的简单估算方法，也许看似合理，但实际上也未必合理，都需要结合系统真实情况（比如是IO密集型或者是CPU密集型或者是纯内存操作）和硬件环境（CPU、内存、硬盘读写速度、网络状况等）来不断尝试达到一个符合实际的合理估算值。

最后来一个“Dark Magic”估算方法（因为我暂时还没有搞懂它的原理），使用下面的类：

001 package pool_size_calculate;
002  
003 import java.math.BigDecimal;
004 import java.math.RoundingMode;
005 import java.util.Timer;
006 import java.util.TimerTask;
007 import java.util.concurrent.BlockingQueue;
008  
009 /**
010  * A class that calculates the optimal thread pool boundaries. It takes the
011  * desired target utilization and the desired work queue memory consumption as
012  * input and retuns thread count and work queue capacity.
013  *
014  * @author Niklas Schlimm
015  *
016  */
017 public abstract class PoolSizeCalculator {
018  
019     /**
020      * The sample queue size to calculate the size of a single {@link Runnable}
021      * element.
022      */
023     private final int SAMPLE_QUEUE_SIZE = 1000;
024  
025     /**
026      * Accuracy of test run. It must finish within 20ms of the testTime
027      * otherwise we retry the test. This could be configurable.
028      */
029     private final int EPSYLON = 20;
030  
031     /**
032      * Control variable for the CPU time investigation.
033      */
034     private volatile boolean expired;
035  
036     /**
037      * Time (millis) of the test run in the CPU time calculation.
038      */
039     private final long testtime = 3000;
040  
041     /**
042      * Calculates the boundaries of a thread pool for a given {@link Runnable}.
043      *
044      * @param targetUtilization
045      *            the desired utilization of the CPUs (0 <= targetUtilization <=   *            1)     * @param targetQueueSizeBytes   *            the desired maximum work queue size of the thread pool (bytes)     */     protected void calculateBoundaries(BigDecimal targetUtilization,            BigDecimal targetQueueSizeBytes) {      calculateOptimalCapacity(targetQueueSizeBytes);         Runnable task = creatTask();        start(task);        start(task); // warm up phase       long cputime = getCurrentThreadCPUTime();       start(task); // test intervall      cputime = getCurrentThreadCPUTime() - cputime;      long waittime = (testtime * 1000000) - cputime;         calculateOptimalThreadCount(cputime, waittime, targetUtilization);  }   private void calculateOptimalCapacity(BigDecimal targetQueueSizeBytes) {        long mem = calculateMemoryUsage();      BigDecimal queueCapacity = targetQueueSizeBytes.divide(new BigDecimal(              mem), RoundingMode.HALF_UP);        System.out.println("Target queue memory usage (bytes): "                + targetQueueSizeBytes);        System.out.println("createTask() produced "                 + creatTask().getClass().getName() + " which took " + mem               + " bytes in a queue");         System.out.println("Formula: " + targetQueueSizeBytes + " / " + mem);       System.out.println("* Recommended queue capacity (bytes): "                 + queueCapacity);   }   /**      * Brian Goetz‘ optimal thread count formula, see ‘Java Concurrency in   * Practice‘ (chapter 8.2)   *       * @param cpu    *            cpu time consumed by considered task   * @param wait   *            wait time of considered task   * @param targetUtilization      *            target utilization of the system   */     private void calculateOptimalThreadCount(long cpu, long wait,           BigDecimal targetUtilization) {         BigDecimal waitTime = new BigDecimal(wait);         BigDecimal computeTime = new BigDecimal(cpu);       BigDecimal numberOfCPU = new BigDecimal(Runtime.getRuntime()                .availableProcessors());        BigDecimal optimalthreadcount = numberOfCPU.multiply(targetUtilization)                 .multiply(                      new BigDecimal(1).add(waitTime.divide(computeTime,                              RoundingMode.HALF_UP)));        System.out.println("Number of CPU: " + numberOfCPU);        System.out.println("Target utilization: " + targetUtilization);         System.out.println("Elapsed time (nanos): " + (testtime * 1000000));        System.out.println("Compute time (nanos): " + cpu);         System.out.println("Wait time (nanos): " + wait);       System.out.println("Formula: " + numberOfCPU + " * "                + targetUtilization + " * (1 + " + waitTime + " / "                 + computeTime + ")");       System.out.println("* Optimal thread count: " + optimalthreadcount);    }   /**      * Runs the {@link Runnable} over a period defined in {@link #testtime}.     * Based on Heinz Kabbutz‘ ideas     * (http://www.javaspecialists.eu/archive/Issue124.html).    *       * @param task   *            the runnable under investigation   */     public void start(Runnable task) {      long start = 0;         int runs = 0;       do {            if (++runs > 5) {
046                 throw new IllegalStateException("Test not accurate");
047             }
048             expired = false;
049             start = System.currentTimeMillis();
050             Timer timer = new Timer();
051             timer.schedule(new TimerTask() {
052                 public void run() {
053                     expired = true;
054                 }
055             }, testtime);
056             while (!expired) {
057                 task.run();
058             }
059             start = System.currentTimeMillis() - start;
060             timer.cancel();
061         } while (Math.abs(start - testtime) > EPSYLON);
062         collectGarbage(3);
063     }
064  
065     private void collectGarbage(int times) {
066         for (int i = 0; i < times; i++) {
067             System.gc();
068             try {
069                 Thread.sleep(10);
070             } catch (InterruptedException e) {
071                 Thread.currentThread().interrupt();
072                 break;
073             }
074         }
075     }
076  
077     /**
078      * Calculates the memory usage of a single element in a work queue. Based on
079      * Heinz Kabbutz‘ ideas
080      * (http://www.javaspecialists.eu/archive/Issue029.html).
081      *
082      * @return memory usage of a single {@link Runnable} element in the thread
083      *         pools work queue
084      */
085     public long calculateMemoryUsage() {
086         BlockingQueue queue = createWorkQueue();
087         for (int i = 0; i < SAMPLE_QUEUE_SIZE; i++) {
088             queue.add(creatTask());
089         }
090         long mem0 = Runtime.getRuntime().totalMemory()
091                 - Runtime.getRuntime().freeMemory();
092         long mem1 = Runtime.getRuntime().totalMemory()
093                 - Runtime.getRuntime().freeMemory();
094         queue = null;
095         collectGarbage(15);
096         mem0 = Runtime.getRuntime().totalMemory()
097                 - Runtime.getRuntime().freeMemory();
098         queue = createWorkQueue();
099         for (int i = 0; i < SAMPLE_QUEUE_SIZE; i++) {
100             queue.add(creatTask());
101         }
102         collectGarbage(15);
103         mem1 = Runtime.getRuntime().totalMemory()
104                 - Runtime.getRuntime().freeMemory();
105         return (mem1 - mem0) / SAMPLE_QUEUE_SIZE;
106     }
107  
108     /**
109      * Create your runnable task here.
110      *
111      * @return an instance of your runnable task under investigation
112      */
113     protected abstract Runnable creatTask();
114  
115     /**
116      * Return an instance of the queue used in the thread pool.
117      *
118      * @return queue instance
119      */
120     protected abstract BlockingQueue createWorkQueue();
121  
122     /**
123      * Calculate current cpu time. Various frameworks may be used here,
124      * depending on the operating system in use. (e.g.
125      * http://www.hyperic.com/products/sigar). The more accurate the CPU time
126      * measurement, the more accurate the results for thread count boundaries.
127      *
128      * @return current cpu time of current thread
129      */
130     protected abstract long getCurrentThreadCPUTime();
131  
132 }

然后自己继承这个抽象类并实现它的三个抽象方法，比如下面是我写的一个示例（任务是请求网络数据），其中我指定期望CPU利用率为1.0（即100%），任务队列总大小不超过100,000字节：

01 package pool_size_calculate;
02  
03 import java.io.BufferedReader;
04 import java.io.IOException;
05 import java.io.InputStreamReader;
06 import java.lang.management.ManagementFactory;
07 import java.math.BigDecimal;
08 import java.net.HttpURLConnection;
09 import java.net.URL;
10 import java.util.concurrent.BlockingQueue;
11 import java.util.concurrent.LinkedBlockingQueue;
12  
13 public class SimplePoolSizeCaculatorImpl extends PoolSizeCalculator {
14  
15     @Override
16     protected Runnable creatTask() {
17         return new AsyncIOTask();
18     }
19  
20     @Override
21     protected BlockingQueue createWorkQueue() {
22         return new LinkedBlockingQueue(1000);
23     }
24  
25     @Override
26     protected long getCurrentThreadCPUTime() {
27         return ManagementFactory.getThreadMXBean().getCurrentThreadCpuTime();
28     }
29  
30     public static void main(String[] args) {
31         PoolSizeCalculator poolSizeCalculator = new SimplePoolSizeCaculatorImpl();
32         poolSizeCalculator.calculateBoundaries(new BigDecimal(1.0), new BigDecimal(100000));
33     }
34  
35 }
36  
37 /**
38  * 自定义的异步IO任务
39  * @author Will
40  *
41  */
42 class AsyncIOTask implements Runnable {
43  
44     @Override
45     public void run() {
46         HttpURLConnection connection = null;
47         BufferedReader reader = null;
48         try {
49             String getURL = "http://baidu.com";
50             URL getUrl = new URL(getURL);
51  
52             connection = (HttpURLConnection) getUrl.openConnection();
53             connection.connect();
54             reader = new BufferedReader(new InputStreamReader(
55                     connection.getInputStream()));
56  
57             String line;
58             while ((line = reader.readLine()) != null) {
59                 // empty loop
60             }
61         }
62  
63         catch (IOException e) {
64  
65         } finally {
66             if(reader != null) {
67                 try {
68                     reader.close();
69                 }
70                 catch(Exception e) {
71  
72                 }
73             }
74             connection.disconnect();
75         }
76  
77     }
78  
79 }

得到的输出如下：

01 Target queue memory usage (bytes): 100000
02 createTask() produced pool_size_calculate.AsyncIOTask which took 40 bytes in a queue
03 Formula: 100000 / 40
04 * Recommended queue capacity (bytes): 2500
05 Number of CPU: 4
06 Target utilization: 1
07 Elapsed time (nanos): 3000000000
08 Compute time (nanos): 47181000
09 Wait time (nanos): 2952819000
10 Formula: 4 * 1 * (1 + 2952819000 / 47181000)
11 * Optimal thread count: 256

推荐的任务队列大小为2500，线程数为256，有点出乎意料之外。我可以如下构造一个线程池：

1 ThreadPoolExecutor pool =
2  new ThreadPoolExecutor(256, 256, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(2500));

原创文章，转载请注明： 转载自并发编程网 – ifeve.com本文链接地址: 如何合理地估算线程池大小？

如何合理地估算线程池大小？

标签：单线程   line   imu   验证   不能   cond   target   试验   puts   

原文地址：https://www.cnblogs.com/tiancai/p/9934767.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！