zoukankan      html  css  js  c++  java
  • 如何合理地估算线程池大小?

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

    这个问题虽然看起来很小,却并不那么容易回答。大家如果有更好的方法欢迎赐教,先来一个天真的估算方法:假设要求一个系统的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总核数):

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

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

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

    最佳线程数目 = ((线程等待时间+线程CPU时间)/线程CPU时间 )* CPU数目

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

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

    可以得出一个结论:线程等待时间所占比例越高,需要越多线程。线程CPU时间所占比例越高,需要越少线程。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    复制代码
      1 package threadpool;
      2 
      3 import java.math.BigDecimal;
      4 import java.math.RoundingMode;
      5 import java.util.Timer;
      6 import java.util.TimerTask;
      7 import java.util.concurrent.BlockingQueue;
      8 
      9 /**
     10  * A class that calculates the optimal thread pool boundaries. It takes the
     11  * desired target utilization and the desired work queue memory consumption as
     12  * input and retuns thread count and work queue capacity.
     13  *
     14  * @author Niklas Schlimm
     15  */
     16 public abstract class PoolSizeCalculator {
     17 
     18     /**
     19      * The sample queue size to calculate the size of a single {@link Runnable}
     20      * element.
     21      */
     22     private final int SAMPLE_QUEUE_SIZE = 1000;
     23 
     24     /**
     25      * Accuracy of test run. It must finish within 20ms of the testTime
     26      * otherwise we retry the test. This could be configurable.
     27      */
     28     private final int EPSYLON = 20;
     29 
     30     /**
     31      * Control variable for the CPU time investigation.
     32      */
     33     private volatile boolean expired;
     34 
     35     /**
     36      * Time (millis) of the test run in the CPU time calculation.
     37      */
     38     private final long testtime = 3000;
     39 
     40     /**
     41      * Calculates the boundaries of a thread pool for a given {@link Runnable}.
     42      *
     43      * @param targetUtilization the desired utilization of the CPUs (0 <= targetUtilization <=      *            1)      * @param targetQueueSizeBytes      *            the desired maximum work queue size of the thread pool (bytes)
     44      */
     45     protected void calculateBoundaries(BigDecimal targetUtilization, BigDecimal targetQueueSizeBytes) {
     46         calculateOptimalCapacity(targetQueueSizeBytes);
     47         Runnable task = creatTask();
     48         start(task);
     49         start(task); // warm up phase
     50         long cputime = getCurrentThreadCPUTime();
     51         start(task); // test intervall
     52         cputime = getCurrentThreadCPUTime() - cputime;
     53         long waittime = (testtime * 1000000) - cputime;
     54         calculateOptimalThreadCount(cputime, waittime, targetUtilization);
     55     }
     56 
     57     private void calculateOptimalCapacity(BigDecimal targetQueueSizeBytes) {
     58         long mem = calculateMemoryUsage();
     59         BigDecimal queueCapacity = targetQueueSizeBytes.divide(new BigDecimal(mem),
     60                 RoundingMode.HALF_UP);
     61         System.out.println("Target queue memory usage (bytes): "
     62                 + targetQueueSizeBytes);
     63         System.out.println("createTask() produced " + creatTask().getClass().getName() + " which took " + mem + " bytes in a queue");
     64         System.out.println("Formula: " + targetQueueSizeBytes + " / " + mem);
     65         System.out.println("* Recommended queue capacity (bytes): " + queueCapacity);
     66     }
     67 
     68     /**
     69      * Brian Goetz' optimal thread count formula, see 'Java Concurrency in
     70      * * Practice' (chapter 8.2)      *
     71      * * @param cpu
     72      * *            cpu time consumed by considered task
     73      * * @param wait
     74      * *            wait time of considered task
     75      * * @param targetUtilization
     76      * *            target utilization of the system
     77      */
     78     private void calculateOptimalThreadCount(long cpu, long wait,
     79                                              BigDecimal targetUtilization) {
     80         BigDecimal waitTime = new BigDecimal(wait);
     81         BigDecimal computeTime = new BigDecimal(cpu);
     82         BigDecimal numberOfCPU = new BigDecimal(Runtime.getRuntime()
     83                 .availableProcessors());
     84         BigDecimal optimalthreadcount = numberOfCPU.multiply(targetUtilization)
     85                 .multiply(new BigDecimal(1).add(waitTime.divide(computeTime,
     86                         RoundingMode.HALF_UP)));
     87         System.out.println("Number of CPU: " + numberOfCPU);
     88         System.out.println("Target utilization: " + targetUtilization);
     89         System.out.println("Elapsed time (nanos): " + (testtime * 1000000));
     90         System.out.println("Compute time (nanos): " + cpu);
     91         System.out.println("Wait time (nanos): " + wait);
     92         System.out.println("Formula: " + numberOfCPU + " * "
     93                 + targetUtilization + " * (1 + " + waitTime + " / "
     94                 + computeTime + ")");
     95         System.out.println("* Optimal thread count: " + optimalthreadcount);
     96     }
     97 
     98     /**
     99      * * Runs the {@link Runnable} over a period defined in {@link #testtime}.
    100      * * Based on Heinz Kabbutz' ideas
    101      * * (http://www.javaspecialists.eu/archive/Issue124.html).
    102      * *
    103      * * @param task
    104      * *            the runnable under investigation
    105      */
    106     public void start(Runnable task) {
    107         long start = 0;
    108         int runs = 0;
    109         do {
    110             if (++runs > 5) {
    111                 throw new IllegalStateException("Test not accurate");
    112             }
    113             expired = false;
    114             start = System.currentTimeMillis();
    115             Timer timer = new Timer();
    116             timer.schedule(new TimerTask() {
    117                 public void run() {
    118                     expired = true;
    119                 }
    120             }, testtime);
    121             while (!expired) {
    122                 task.run();
    123             }
    124             start = System.currentTimeMillis() - start;
    125             timer.cancel();
    126         } while (Math.abs(start - testtime) > EPSYLON);
    127         collectGarbage(3);
    128     }
    129 
    130     private void collectGarbage(int times) {
    131         for (int i = 0; i < times; i++) {
    132             System.gc();
    133             try {
    134                 Thread.sleep(10);
    135             } catch (InterruptedException e) {
    136                 Thread.currentThread().interrupt();
    137                 break;
    138             }
    139         }
    140     }
    141 
    142     /**
    143      * Calculates the memory usage of a single element in a work queue. Based on
    144      * Heinz Kabbutz' ideas
    145      * (http://www.javaspecialists.eu/archive/Issue029.html).
    146      *
    147      * @return memory usage of a single {@link Runnable} element in the thread
    148      * pools work queue
    149      */
    150     public long calculateMemoryUsage() {
    151         BlockingQueue queue = createWorkQueue();
    152         for (int i = 0; i < SAMPLE_QUEUE_SIZE; i++) {
    153             queue.add(creatTask());
    154         }
    155 
    156         long mem0 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
    157         long mem1 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
    158 
    159         queue = null;
    160 
    161         collectGarbage(15);
    162 
    163         mem0 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
    164         queue = createWorkQueue();
    165 
    166         for (int i = 0; i < SAMPLE_QUEUE_SIZE; i++) {
    167             queue.add(creatTask());
    168         }
    169 
    170         collectGarbage(15);
    171 
    172         mem1 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();
    173 
    174         return (mem1 - mem0) / SAMPLE_QUEUE_SIZE;
    175     }
    176 
    177     /**
    178      * Create your runnable task here.
    179      *
    180      * @return an instance of your runnable task under investigation
    181      */
    182     protected abstract Runnable creatTask();
    183 
    184     /**
    185      * Return an instance of the queue used in the thread pool.
    186      *
    187      * @return queue instance
    188      */
    189     protected abstract BlockingQueue createWorkQueue();
    190 
    191     /**
    192      * Calculate current cpu time. Various frameworks may be used here,
    193      * depending on the operating system in use. (e.g.
    194      * http://www.hyperic.com/products/sigar). The more accurate the CPU time
    195      * measurement, the more accurate the results for thread count boundaries.
    196      *
    197      * @return current cpu time of current thread
    198      */
    199     protected abstract long getCurrentThreadCPUTime();
    200 
    201 }
    复制代码

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

    复制代码
     1 package threadpool;
     2 
     3 import java.io.BufferedReader;
     4 import java.io.IOException;
     5 import java.io.InputStreamReader;
     6 import java.lang.management.ManagementFactory;
     7 import java.math.BigDecimal;
     8 import java.net.HttpURLConnection;
     9 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     public void run() {
    45         HttpURLConnection connection = null;
    46         BufferedReader reader = null;
    47         try {
    48             String getURL = "http://baidu.com";
    49             URL getUrl = new URL(getURL);
    50 
    51             connection = (HttpURLConnection) getUrl.openConnection();
    52             connection.connect();
    53             reader = new BufferedReader(new InputStreamReader(
    54                     connection.getInputStream()));
    55 
    56             String line;
    57             while ((line = reader.readLine()) != null) {
    58                 // empty loop
    59             }
    60         }
    61 
    62         catch (IOException e) {
    63 
    64         } finally {
    65             if(reader != null) {
    66                 try {
    67                     reader.close();
    68                 }
    69                 catch(Exception e) {
    70 
    71                 }
    72             }
    73             connection.disconnect();
    74         }
    75 
    76     }
    77 
    78 }
    复制代码

    得到如下输出:

    复制代码
    Target queue memory usage (bytes): 100000
    createTask() produced threadpool.AsyncIOTask which took 40 bytes in a queue
    Formula: 100000 / 40
    * Recommended queue capacity (bytes): 2500
    Number of CPU: 8
    Target utilization: 1
    Elapsed time (nanos): 3000000000
    Compute time (nanos): 280801800
    Wait time (nanos): 2719198200
    Formula: 8 * 1 * (1 + 2719198200 / 280801800)
    * Optimal thread count: 88
    复制代码

    推荐的任务队列大小为2500,线程数为88。依次为依据,我们就可以构造这样一个线程池:

    ThreadPoolExecutor pool = new ThreadPoolExecutor(88, 88, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(2500));

    可以将这个文件打包成可执行的jar文件,这样就可以拷贝到测试/正式环境上执行。

    复制代码
     1 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     2   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     3     <modelVersion>4.0.0</modelVersion>
     4 
     5     <groupId>threadpool</groupId>
     6     <artifactId>dark-magic</artifactId>
     7     <version>1.0-SNAPSHOT</version>
     8     <packaging>jar</packaging>
     9 
    10     <name>dark_magic</name>
    11     <url>http://maven.apache.org</url>
    12 
    13     <properties>
    14         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    15     </properties>
    16 
    17     <dependencies>
    18         
    19     </dependencies>
    20 
    21     <build>
    22         <finalName>dark-magic</finalName>
    23 
    24         <plugins>
    25             <plugin>
    26                 <artifactId>maven-assembly-plugin</artifactId>
    27                 <configuration>
    28                     <appendAssemblyId>false</appendAssemblyId>
    29                     <descriptorRefs>
    30                         <descriptorRef>jar-with-dependencies</descriptorRef>
    31                     </descriptorRefs>
    32                     <archive>
    33                         <manifest>
    34                             <!-- 此处指定main方法入口的class -->
    35                             <mainClass>threadpool.SimplePoolSizeCaculatorImpl</mainClass>
    36                         </manifest>
    37                     </archive>
    38                 </configuration>
    39                 <executions>
    40                     <execution>
    41                         <id>make-assembly</id>
    42                         <phase>package</phase>
    43                         <goals>
    44                             <goal>assembly</goal>
    45                         </goals>
    46                     </execution>
    47                 </executions>
    48             </plugin>
    49         </plugins>
    50     </build>
    51 </project>
    复制代码

    转载:

    http://ifeve.com/how-to-calculate-threadpool-size/

    http://www.importnew.com/17384.html

    https://www.cnblogs.com/cherish010/p/8334952.html

  • 相关阅读:
    2020-2021第一学期2024"DCDD"小组第十二周讨论
    2020-2021第一学期《网络空间安全导论》第十二周自习总结
    2020-2021第一学期2024"DCDD"小组第十一周讨论
    2020-2021第一学期《网络空间安全导论》第十一周自习总结
    2020-2021第一学期2024"DCDD"小组第十周讨论
    2019-2020-1 20165213 20165224 20165311 《信息安全系统设计基础》实验五 通讯协议设计
    2019-2020-1 实验三 并发程序
    2019-2020-1 20165213 20165224 20165311 实验二
    2019-2020-1 20165213 20165224 20165311 实验一 开发环境的熟悉
    20165213 Exp9 Web安全基础
  • 原文地址:https://www.cnblogs.com/mrchenzheng/p/12166338.html
Copyright © 2011-2022 走看看