zoukankan      html  css  js  c++  java

  • java线程池分析和应用

    比较

       在前面的一些文章里，我们已经讨论了手工创建和管理线程。在实际应用中我们有的时候也会经常听到线程池这个概念。在这里，我们可以先针对手工创建管理线程和通过线程池来管理做一个比较。通常，我们如果手工创建线程，需要定义线程执行对象，它实现的接口。然后再创建一个线程对象，将我们定义好的对象执行部分装载到线程中。对于线程的创建、结束和结果的获取都需要我们来考虑。如果我们需要用到很多的线程时，对线程的管理就会变得比较困难。我们手工定义线程的方式在时间和空间效率方面会存在着一些不足。比如说我们定义好的线程不能再进行重复利用，以后每次使用线程的时候都需要去向系统申请资源来创建一个。这样会导致对资源利用效率比较低。另外，我们知道线程的创建和启动都需要一定的时间，每次都从头启动一个线程这样效率也比较低。

        以上这些问题恰恰都是线程池所能解决的。笼统的从概念上来说，线程池通过一个缓存池的空间预先创建了一部分线程，在我们需要使用的时候就从里面直接将线程资源取出来使用。在将线程使用完毕之后，线程池又会将线程回收进行再利用。正是因为线程池的这些优点，在一些对性能要求比较高以及线程请求比较多的时候，它是一个很理想的选择，也值得我们好好的研究研究。

    Java线程池

    线程池类型

        自从Java 1.5以来提供的线程池功能，我们使用线程池还是很方便的。一般都是通过Executors类提供的方法来创建。Executors提供了创建一下几类线程池的方法：

    • Single Thread Executor: 创建的线程只包含一个线程，所有提交到线程池的线程会按照提交的顺序一个接一个的执行。通过Executors.newSingleThreadExecutor()方法创建。这种线程池适用于我们只希望每次使用一个线程的情况。
    • Cached Thread Pool: 线程池里会创建尽可能多的必须线程来并行执行。一旦前面的线程执行结束后可以被重复使用。当然，使用这种线程池的时候我们必须要小心。我们使用多线程的目的是希望能够提高并行度和效率，但是并不是线程越多就越好。如果我们设定的线程数目过多的时候，使用Cached Thread Pool并不是一个很理想的选择。因为一方面它占用了大量的线程资源，同时线程之间互相切换很频繁的时候也会带来执行效率的下降。它主要适用于使用的线程数目不多，但是对线程有比较灵活动态的要求。一般通过Executors.newCachedThreadPool()来创建。
    • Fix Thread Pool: 线程池里会创建固定数量的线程。在线程都被使用之后，后续申请使用的线程都会被阻塞在那里。使用Executors.newScheduledThreadPool()创建。
    • Scheduled Thread Pool: 线程池可以对线程执行的时间和顺序做预先指定。比如说要某些线程在某个时候才启动或者每隔多长时间就启动一次。有点像我们的Timer Job。使用Executors.newScheduledThreadPool()创建。
    • Single Thread Scheduled Pool: 线程池按照指定的时间或顺序来启动线程池。同时线程池里只有一个线程。创建方法：Executors.newSingleThreadScheduledExecutor()

        前面的这几种类型的线程池已经能够满足我们大多数的要求。从表面上看，这些线程的创建都是通过Executors的静态方法实现。实际上，我们所有创建的线程池都可以说是ExecutorService类型的。前面创建的线程池比如说Cached Thread Pool都是ExecutorService类型的子类。Executors, ExecutorService和ThreadPoolExecutor等各种具体的线程池的类关系图如下：

         从这个类图我们可以看到，Executors本身会引用到各种具体ExecutorService的实现。它本身相当于是一个定义的工厂方法，将各种具体线程池的创建给封装起来。当然，我们也可以通过实例化具体的类来建立线程池，只不过相对来说更加麻烦。更加推荐使用Executors的工厂方法。

    示例

        有了前面那些讨论，我们可以举一个具体的实例来看看线程池的使用。通常来说，我们使用线程可以分为两种类型。一种是使用多个线程执行某些任务，但不一定要将线程执行的结果统一返回并统计。另外一种则需要将线程执行的结果统一记录和统计。我们就针对这两种情况来尝试。

    运行线程不统一返回结果

        假定我们就通过线程池创建若干个线程，每个线程仅仅是休眠若干秒钟然后打印一些信息。我们可以这样来实现代码：

    首先定义要执行的线程：

    Java代码  收藏代码
    1. import java.util.Date;  
    2. import java.util.concurrent.TimeUnit;  
    3.   
    4. public class Task implements Runnable {  
    5.     private Date initDate;  
    6.     private String name;  
    7.       
    8.     public Task(String name) {  
    9.         initDate = new Date();  
    10.         this.name = name;  
    11.     }  
    12.       
    13.     @Override  
    14.     public void run() {  
    15.         System.out.printf("%s: Task %s: Created on %s ",  
    16.                 Thread.currentThread().getName(), name, initDate);  
    17.         System.out.printf("%s: Task %s: Started on %s ",  
    18.                 Thread.currentThread().getName(), name, new Date());  
    19.         try {  
    20.             long duration = (long)(Math.random() * 10);  
    21.             System.out.printf("%s: Task %s: Doing a task during %d seconds ",  
    22.                     Thread.currentThread().getName(), name, duration);  
    23.             TimeUnit.SECONDS.sleep(duration);  
    24.         } catch(InterruptedException e) {  
    25.             e.printStackTrace();  
    26.         }  
    27.           
    28.         System.out.printf("%s: Task %s: Finished on: %s ",  
    29.                 Thread.currentThread().getName(), name, new Date());  
    30.     }  
    31. }  

         这部分代码很简单，就是定义一个Task类，它实现Runnable接口。

        然后我们再定义一个封装了线程池的类：

    Java代码  收藏代码
    1. import java.util.concurrent.Executors;  
    2. import java.util.concurrent.ThreadPoolExecutor;  
    3.   
    4. public class Server {  
    5.     private ThreadPoolExecutor executor;  
    6.       
    7.     public Server() {  
    8.         executor = (ThreadPoolExecutor)Executors.newCachedThreadPool();  
    9.     }  
    10.       
    11.     public void executeTask(Task task) {  
    12.         System.out.printf("Server: A new task has arrived ");  
    13.         executor.execute(task);  
    14.         System.out.printf("Server: Pool Size: %d ", executor.getPoolSize());  
    15.         System.out.printf("Server: Active Count: %d ", executor.getActiveCount());  
    16.         System.out.printf("Server: Completed Tasks: %d ", executor.getCompletedTaskCount());  
    17.     }  
    18.       
    19.     public void endServer() {  
    20.         executor.shutdown();  
    21.     }  
    22. }  

         Server类主要是内部定义了ThreadPoolExecutor的成员变量和executeTask, endServer两个方法。他们一个用于分配线程并执行，另外一个用于关闭整个线程池。

        我们整个使用的代码如下：

    Java代码  收藏代码
    1. public class Main {  
    2.   
    3.     public static void main(String[] args) {  
    4.         Server server = new Server();  
    5.         for(int i = 0; i < 10; i++) {  
    6.             Task task = new Task("Task " + i);  
    7.             server.executeTask(task);  
    8.         }  
    9.         server.endServer();  
    10.     }  
    11. }  

         在这里我们通过线程池创建了10个线程。在执行结束之后我们要注意的一点是必须要调用endServer来终止线程池，否则线程池会一直处在一个运行的状态。

        总的来说，前面使用线程池的方法非常简单。无非是创建线程执行对象，再将它传入给线程池的execute方法。最后，在使用完毕线程池之后关闭它。

    运行线程统一返回结果

        现在假定我们要使用一个线程池来调度几个线程。每个线程来计算一个 给定 数字的阶乘n!。那么，我们可以先定义需要提交计算的部分：

    Java代码  收藏代码
    1. import java.util.concurrent.Callable;  
    2. import java.util.concurrent.TimeUnit;  
    3.   
    4. public class FactorialCalculator implements Callable<Integer> {  
    5.     private Integer number;  
    6.       
    7.     public FactorialCalculator(Integer number) {  
    8.         this.number = number;  
    9.     }  
    10.       
    11.     @Override  
    12.     public Integer call() throws Exception {  
    13.         int result = 1;  
    14.           
    15.         if(number == 0 || number == 1) {  
    16.             result = 1;  
    17.         } else {  
    18.             for(int i = 2; i <= number; i++) {  
    19.                 result *= i;  
    20.                 TimeUnit.MILLISECONDS.sleep(200);  
    21.             }  
    22.         }  
    23.         System.out.printf("%s: %d ", Thread.currentThread().getName(), result);  
    24.         return result;  
    25.     }  
    26. }  

         这里，我们定义了FactorialCalculator类。它实现了接口Callable。这个接口后面会被ThreadPoolExecutor的submit方法所用到。线程池在收到Callable的参数后会启动一个线程来执行call方法里的运算。还有一个需要注意到的地方是，我们这里定义了返回值是Integer类型的。

        现在我们再来看怎么使用他们：

    Java代码  收藏代码
    1. import java.util.ArrayList;  
    2. import java.util.List;  
    3. import java.util.Random;  
    4. import java.util.concurrent.ExecutionException;  
    5. import java.util.concurrent.Executors;  
    6. import java.util.concurrent.Future;  
    7. import java.util.concurrent.ThreadPoolExecutor;  
    8.   
    9.   
    10. public class Main {  
    11.   
    12.     public static void main(String[] args) {  
    13.         //创建固定长度为2的线程池  
    14.                 ThreadPoolExecutor executor = (ThreadPoolExecutor)Executors.  
    15.                 newFixedThreadPool(2);  
    16.                 // 声明保存返回结果的列表，注意类型为Future<Integer>  
    17.         List<Future<Integer>> resultList = new ArrayList<>();  
    18.         Random random = new Random();  
    19.   
    20.                 // For循环中的submit方法在提交线程执行后会有一个返回类型为Future<Integer>的结果。将结果保存在列表中。  
    21.         for(int i = 0; i < 10; i++) {  
    22.             Integer number = random.nextInt(10);  
    23.             FactorialCalculator calculator =   
    24.                     new FactorialCalculator(number);  
    25.             Future<Integer> result = executor.submit(calculator);  
    26.             resultList.add(result);  
    27.         }  
    28.   
    29.         System.out.printf("Main: Results ");  
    30.         for(int i = 0; i < resultList.size(); i++) {  
    31.             Future<Integer> result = resultList.get(i);  
    32.             Integer number = null;  
    33.             try {  
    34.                                 // 结果需要在线程执行完后才能get到，所以get执行时会使得线程等待，需要捕捉异常  
    35.                 number = result.get();  
    36.             } catch(InterruptedException e) {  
    37.                 e.printStackTrace();  
    38.             } catch(ExecutionException e) {  
    39.                 e.printStackTrace();  
    40.             }  
    41.             System.out.printf("Main: Task %d: %d ", i, number);  
    42.         }  
    43.           
    44.                 // 关闭线程池  
    45.         executor.shutdown();  
    46.     }  
    47. }  

         前面的代码我增加了一些说明的注释。其中的要点就是我们submit提交运算之后并不会马上得到结果。但是这个结果集我们可以将他们保存到一个列表中。在后面通过get的方法来获取。

        这里还有一个比较有意思的地方就是，在前面那种不返回结果的地方，我们是通过提交一个Runnable的参数给ThreadPoolExecutor的execute方法来安排线程执行。这个我们好理解。可是在后面这个返回结果的地方，我们是提交了一个Callable的参数给ThreadPoolExecutor的submit方法。这里没有声明线程的东西，比如说我实现Runnable或者集成Thread，我怎么知道分配线程来执行call方法里面的东西呢？ 这个其实在ThreadPoolExecutor继承的类结构里有一个转换的方式，将Callable的变量绑定到一个线程对象实例当中，然后启动线程执行。在后续线程池详细的实现分析中我们再深入讨论。

        综合这部分的讨论，返回结果集其实也就那么回事。不就是定义一个实现Callable的对象提交给线程池的submit方法么？后面我们只要等着去读submit返回的Future<T>结果就行了。So easy!

    Java线程池和线程数的选择

        我们使用线程池来做多线程运算的时候，问题类型和对应线程数量的选择都很重要。一般来说，我们面临的多线程处理问题会分为如下两类：

    CPU密集

        对于CPU密集类型的问题来说，更多只是CPU要做大量的运算处理。这个时候如果要保证系统最充分的利用率，我们最好定义的线程数量和系统的CPU数量一致。这样每个CPU都可以充分的运用起来，而且很少或者不会有线程之间的切换。在这种情况下，比较理想的线程数目是CPU的个数或者比CPU个数大1。一般来说，我们可以通过如下的代码来得到CPU的个数：

    Java代码  收藏代码
    1. Runtime.getRuntime().availableProcessors();  

    IO密集

        对于IO密集型的问题来说，我们会发现由于IO经常会带来各种中断，并且IO的速度和CPU处理速度差别很大。因此，我们需要考虑到IO在整个运算时间中所占的比例。比如说我们有50%比例的时间是阻塞的，那么我们可以创建相当于当前CPU数目的两倍数的线程。假设我们知道阻塞的比例数的话，我们可以通过如下的一个简单关系来估算线程数的大小：

    线程数 = 可用CPU个数 / （1 - 阻塞比例数）

        当然，这种估算的方式是基于一个理想的条件，在实际问题中还需要根据具体场景来调整。

    总结

         和手动创建和管理线程比起来，使用线程池来管理线程是一种更加理想的选择。我们可以根据具体问题对线程的要求来选择不同的线程池。创建线程池一般推荐使用Executors里定义的工厂方法，它本身屏蔽了很多创建线程池的细节。另外，我们通常使用线程池来启用多个线程的时候，有的时候不需要去刻意统计每个线程执行的结果，有的时候又需要。可以笼统的将线程池的使用分成这两类。在我们不需要统计结果的时候可以直接用线程池里ThreadPoolExecutor的execute(Runnable command)方法来执行线程。如果我们需要获得线程执行的结果，可以使用ThreadPoolExecutor的submit(Callable call)来提交线程，然后通过收集submit返回的Future<T>结果进行统计。

        另外，我们选择线程池执行多任务的时候也需要考虑执行的任务类型。通常这些任务可以分为CPU密集和IO密集的两种类型。对于CPU密集的任务，我们应该选择和CPU数量一致的线程数，这样可以达到理想的执行效果也减少了线程切换的开销。对于IO密集的任务，我们可以根据IO所占程序执行中的比例来选择，一般推荐线程数目是CPU数目的10倍。
  • 相关阅读:
    第四章、数值计算
    一、bif
    十三、LaTex中的参考文献BibTex
    图像分类数据组织方式
    深度学习中loss总结
    类别不平衡问题
    各种优化器的比较
    机器学习优化器总结
    空洞卷积
    深度学习之语义分割中的度量标准
  • 原文地址：https://www.cnblogs.com/googlemeoften/p/5769277.html
Copyright © 2011-2022 走看看