zoukankan      html  css  js  c++  java
  • 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倍。

    转自:http://shmilyaw-hotmail-com.iteye.com/blog/1765751

  • 相关阅读:
    几种常用的曲线
    0188. Best Time to Buy and Sell Stock IV (H)
    0074. Search a 2D Matrix (M)
    0189. Rotate Array (E)
    0148. Sort List (M)
    0859. Buddy Strings (E)
    0316. Remove Duplicate Letters (M)
    0452. Minimum Number of Arrows to Burst Balloons (M)
    0449. Serialize and Deserialize BST (M)
    0704. Binary Search (E)
  • 原文地址:https://www.cnblogs.com/harry335/p/5099391.html
Copyright © 2011-2022 走看看