比较
在前面的一些文章里,我们已经讨论了手工创建和管理线程。在实际应用中我们有的时候也会经常听到线程池这个概念。在这里,我们可以先针对手工创建管理线程和通过线程池来管理做一个比较。通常,我们如果手工创建线程,需要定义线程执行对象,它实现的接口。然后再创建一个线程对象,将我们定义好的对象执行部分装载到线程中。对于线程的创建、结束和结果的获取都需要我们来考虑。如果我们需要用到很多的线程时,对线程的管理就会变得比较困难。我们手工定义线程的方式在时间和空间效率方面会存在着一些不足。比如说我们定义好的线程不能再进行重复利用,以后每次使用线程的时候都需要去向系统申请资源来创建一个。这样会导致对资源利用效率比较低。另外,我们知道线程的创建和启动都需要一定的时间,每次都从头启动一个线程这样效率也比较低。
以上这些问题恰恰都是线程池所能解决的。笼统的从概念上来说,线程池通过一个缓存池的空间预先创建了一部分线程,在我们需要使用的时候就从里面直接将线程资源取出来使用。在将线程使用完毕之后,线程池又会将线程回收进行再利用。正是因为线程池的这些优点,在一些对性能要求比较高以及线程请求比较多的时候,它是一个很理想的选择,也值得我们好好的研究研究。
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的工厂方法。
示例
有了前面那些讨论,我们可以举一个具体的实例来看看线程池的使用。通常来说,我们使用线程可以分为两种类型。一种是使用多个线程执行某些任务,但不一定要将线程执行的结果统一返回并统计。另外一种则需要将线程执行的结果统一记录和统计。我们就针对这两种情况来尝试。
运行线程不统一返回结果
假定我们就通过线程池创建若干个线程,每个线程仅仅是休眠若干秒钟然后打印一些信息。我们可以这样来实现代码:
首先定义要执行的线程:
import java.util.Date; import java.util.concurrent.TimeUnit; public class Task implements Runnable { private Date initDate; private String name; public Task(String name) { initDate = new Date(); this.name = name; } @Override public void run() { System.out.printf("%s: Task %s: Created on %s\n", Thread.currentThread().getName(), name, initDate); System.out.printf("%s: Task %s: Started on %s\n", Thread.currentThread().getName(), name, new Date()); try { long duration = (long)(Math.random() * 10); System.out.printf("%s: Task %s: Doing a task during %d seconds\n", Thread.currentThread().getName(), name, duration); TimeUnit.SECONDS.sleep(duration); } catch(InterruptedException e) { e.printStackTrace(); } System.out.printf("%s: Task %s: Finished on: %s\n", Thread.currentThread().getName(), name, new Date()); } }
这部分代码很简单,就是定义一个Task类,它实现Runnable接口。
然后我们再定义一个封装了线程池的类:
import java.util.concurrent.Executors; import java.util.concurrent.ThreadPoolExecutor; public class Server { private ThreadPoolExecutor executor; public Server() { executor = (ThreadPoolExecutor)Executors.newCachedThreadPool(); } public void executeTask(Task task) { System.out.printf("Server: A new task has arrived\n"); executor.execute(task); System.out.printf("Server: Pool Size: %d\n", executor.getPoolSize()); System.out.printf("Server: Active Count: %d\n", executor.getActiveCount()); System.out.printf("Server: Completed Tasks: %d\n", executor.getCompletedTaskCount()); } public void endServer() { executor.shutdown(); } }
Server类主要是内部定义了ThreadPoolExecutor的成员变量和executeTask, endServer两个方法。他们一个用于分配线程并执行,另外一个用于关闭整个线程池。
我们整个使用的代码如下:
public class Main { public static void main(String[] args) { Server server = new Server(); for(int i = 0; i < 10; i++) { Task task = new Task("Task " + i); server.executeTask(task); } server.endServer(); } }
在这里我们通过线程池创建了10个线程。在执行结束之后我们要注意的一点是必须要调用endServer来终止线程池,否则线程池会一直处在一个运行的状态。
总的来说,前面使用线程池的方法非常简单。无非是创建线程执行对象,再将它传入给线程池的execute方法。最后,在使用完毕线程池之后关闭它。
运行线程统一返回结果
现在假定我们要使用一个线程池来调度几个线程。每个线程来计算一个 给定 数字的阶乘n!。那么,我们可以先定义需要提交计算的部分:
import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; public class FactorialCalculator implements Callable<Integer> { private Integer number; public FactorialCalculator(Integer number) { this.number = number; } @Override public Integer call() throws Exception { int result = 1; if(number == 0 || number == 1) { result = 1; } else { for(int i = 2; i <= number; i++) { result *= i; TimeUnit.MILLISECONDS.sleep(200); } } System.out.printf("%s: %d\n", Thread.currentThread().getName(), result); return result; } }
这里,我们定义了FactorialCalculator类。它实现了接口Callable。这个接口后面会被ThreadPoolExecutor的submit方法所用到。线程池在收到Callable的参数后会启动一个线程来执行call方法里的运算。还有一个需要注意到的地方是,我们这里定义了返回值是Integer类型的。
现在我们再来看怎么使用他们:
import java.util.ArrayList; import java.util.List; import java.util.Random; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ThreadPoolExecutor; public class Main { public static void main(String[] args) { //创建固定长度为2的线程池 ThreadPoolExecutor executor = (ThreadPoolExecutor)Executors. newFixedThreadPool(2); // 声明保存返回结果的列表,注意类型为Future<Integer> List<Future<Integer>> resultList = new ArrayList<>(); Random random = new Random(); // For循环中的submit方法在提交线程执行后会有一个返回类型为Future<Integer>的结果。将结果保存在列表中。 for(int i = 0; i < 10; i++) { Integer number = random.nextInt(10); FactorialCalculator calculator = new FactorialCalculator(number); Future<Integer> result = executor.submit(calculator); resultList.add(result); } System.out.printf("Main: Results\n"); for(int i = 0; i < resultList.size(); i++) { Future<Integer> result = resultList.get(i); Integer number = null; try { // 结果需要在线程执行完后才能get到,所以get执行时会使得线程等待,需要捕捉异常 number = result.get(); } catch(InterruptedException e) { e.printStackTrace(); } catch(ExecutionException e) { e.printStackTrace(); } System.out.printf("Main: Task %d: %d\n", i, number); } // 关闭线程池 executor.shutdown(); } }
前面的代码我增加了一些说明的注释。其中的要点就是我们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的个数:
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倍。
参考材料
Programming Concurrency on the JVM: Mastering Synchronization, STM, and Actors
http://www.baptiste-wicht.com/2010/09/java-concurrency-part-7-executors-and-thread-pools/
相关推荐
Reference: 《创建Java线程池》[1],《Java线程:新特征-线程池》[2], 《Java线程池学习》[3],《线程池ThreadPoolExecutor使用简介》[4],《Java5中的线程池实例讲解》[5],《ThreadPoolExecutor使用和思考》[6] ...
为了防止资源不足,服务器应用程序需要采取一些办法来限制任何给定时刻处理的请求数目,尽可能减少创建和销毁线程的次数,特别是一些资源耗费比较大的线程的创建和销毁,尽量利用已有对象来进行服务,这就是“池化...
主要介绍了Java线程池的应用,结合具体实例形式分析了java线程池的斐波那契数列计算与模拟工人做工等应用的操作技巧,需要的朋友可以参考下
《Java并发编程从入门到精通》内容包括并发编程概念,线程,线程安全,线程集合类,线程阀,线程池,Fork/Join,线程、线程池在互联网项目开发的应用,线程监控及线程分析,Android中线程应用。 本书适合Java开发...
2.实战2.1通过线程池代码创建线程2.1.Executors源码分析newFixedThreadPoolnewWorkStealingPool(int parallelism)newSingleThreadExecutor()newCachedThreadPool()newSingleThreadScheduledExecutor()...
展示了如何提升Java应用性能,并且给出了优化前与优化后的Java应用程序的性能差别,以实际的实例与数字告诉你,为什么不可以这么做,应该怎么做,深入分析了影响Java应用程序性能的根本原因。本书不是教你怎样使用...
2. 性能优化:学习分析和调优Java应用程序的性能,了解内存管理、垃圾回收、性能监测和调优工具。 3. 反射和字节码操作:学习使用Java的反射机制和字节码操作库,实现动态代理、代码生成和运行时修改类的功能。 4....
*关键技术:系统使用了线程池、消息队列、数据库连接池、socket连接池等技术来保证系统的运行效率以支撑较大的应用需求。 功能介绍 *支持通过JAVA和WEB脚本进行业务逻辑开发。 *灵活的业务路有功能可根据命令字和...
1.6 单独编译和打包Android应用程序模块 1.6.1 导入单独编译模块的mmm命令 1.6.2 单独编译Android应用程序模块 1.6.3 重新打包Android系统镜像文件 第2章 硬件抽象层 2.1 开发Android硬件驱动程序 ...
java7在并发编程方面,带来了很多令人激动的新功能,这将使你的应用程序具备更好的并行任务性能。 《Java 7并发编程实战手册》是Java 7并发编程的实战指南,介绍了Java 7并发API中大部分重要而有用的机制。全书分为9...
实时线程栈监控、线程池监控、动态arthas命令集、依赖分析等等等,助你快速定位问题。兼容性:整体监控不管你是IDC、ECS、Docker部署,都可完美兼容。易用: 无需任何配置,开箱即用,基于agent无侵入接入,升级应用...
前端 1 移动端UI一致性解决方案 1 美团外卖Flutter动态化实践 26 美团开源 Logan Web:前端日志在 ...Java线程池实现原理及其在美团业务中的实践 245 美团万亿级 KV 存储架构与实践 276 Java中9种常见的CMS GC问题分析
1.3.2 Java程序的运行机制和JVM 6 1.4 开发Java的准备 7 1.4.1 安装JDK 8 学生提问:不是说JVM是运行Java程序的虚拟机吗?那JRE和JVM的关系是怎样的呢? 8 学生提问:为什么不安装公共JRE系统呢? 9 1.4.2 设置...
流程控制语句,和算法应用 函数式编程概念和应用 集合容器 基于分析列表集合源码体系 基于分析地图集合源码体系 IO流机制 IO流核心模块及基本原理 曼编程 线程的创建方式与生命周期 线程核心机制,基础概念扩展 ...
各领域数据集,工具源码,...云计算与大数据:数据集、包括云计算平台、大数据分析、人工智能、机器学习等,云计算是一种基于互联网的计算方式,通过这种方式,共享的软硬件资源和信息可以按需提供给计算机和其他设备。
1.6 单独编译和打包Android应用程序模块 1.6.1 导入单独编译模块的mmm命令 1.6.2 单独编译Android应用程序模块 1.6.3 重新打包Android系统镜像文件 第2章 硬件抽象层 2.1 开发Android硬件驱动程序 2.1.1 ...
某些应用程序中存在比较明显的任务边界,而在其他一些程序中则需要进一步分析才能揭示出粒度更细的并行性 任务的取消和关闭 任务取消 停止基于线程的服务 处理非正常的线程终止 JVM关闭 线程池...
Vert.x是一个在JVM开发reactive应用的框架,可用于开发异步、可伸缩、高并发的Web...本文试图揭示Vert.x的线程模型的应用,通过源代码分析Vert.x如何使用线程池处理请求的,以及比较Netty和Vert.x使用线程池的异同。
3.2 原理分析 3.3 JVM中锁的优化 3.4 Synchronized与Lock 3.5 扩展 四、volatile 详解 4.1 作用 4.2 实现原理 4.3 应用场景 五、final 详解 5.1 基础 5.2 重排序规则 5.3 扩展 六、JUC 6.1 汇总 6.2 ...