在实际生产环境中,线程的数量必须得到控制.大量创建线程对系统性能是有伤害的.
为了避免系统频繁的创建和销毁线程,我们可以让创建的线程进行复用,线程池中,总有那么几个活跃线程,当需要使用线程时,可以从池子中随便拿一个空闲线程,当完成工作时,不是马上关闭线程,而是将这个线程放回线程池,方便下次再用.
通过Executors 可以创建特定功能的线程池,以下是平时主要用的几种:
public static ExecutorService newFixedThreadPool(int nThreads)
public static ExecutorService newSingleThreadExecutor()
public static ExecutorService newCachedThreadPool()
- newFixedThreadPool(int nThreads)方法
/**
* @param nThreads the number of threads in the pool
* @return the newly created thread pool
* @throws IllegalArgumentException if {@code nThreads <= 0}
*/
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
根据JDK里面注释可以得到: 该方法返回一个固定线程数量的线程池,
当有新的任务提交,线程池中如果有空闲线程,则立即执行.如果所有线程都在活动状态,则新的任务会被暂存的在一个任务队列中,等到线程空闲时,就处理任务队列中的任务.
- newSingleThreadExecutor()方法
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
改方法传回一个只有一个线程的线程池,多余的任务提交后将会保存到任务队列中,待线程空闲,按先入先出的顺序执行队列中的任务.
- newCachedThreadPool()方法
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
改方法返回一个可根据实际情况调整线程数量的线程池.线程池的线程数量不确定,如果有空闲线程,就用空闲线程,如果所有线程都在工作,又有新的任务提交,则会创建新的线程处理任务.
简单介绍线程池的用法
public class ThreadPoolDemo {
public static class Mytask implements Runnable{
@Override
public void run() {
System.out.println(System.currentTimeMillis()+ "---" + Thread.currentThread().getId());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
Mytask mytask = new Mytask();
ExecutorService service = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
service.submit(mytask);
}
}
}
得到运行结果
根据结果可以看到
10个任务分为两批执行,第一批和第二批的线程ID正好是一致的,而且前五个和后五个任务的时间正好相差一秒钟.
- 线程池的内部实现:
根据上面三个简单线程池的代码表面分析,都是使用ThreadPoolExecutor实现.现在分析一下ThreadPoolExecutor;
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param threadFactory the factory to use when the executor
* creates a new thread
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
- corePoolSize : 线程池中的线程数量
- maximumPoolSize : 线程池中可以容纳的最大线程数
- keepAliveTime : 超过指定corePoolSize 的线程的存活时间.
- unit : keepAliveTime的时间单位
- workQueue : 没有被提交的任务所存放的任务队列.
- ThreadFactory: 线程工厂
- RejectedExecutionHandler : 拒绝策略, 当任务来不及处理,如何拒绝任务.
workQueue 没有被提交的任务所存放的任务队列 ,它是BlockingQueue接口的具体实现,下面是对不同线程池的不同分析
通过看源码可以得到:
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
newFixedThreadPool 该线程池的开始线程数量和最大线程数量是一开始就设置好的,并且使用了LinkedBlockingQueue 存放没有被执行的线程, LinkedBlockingQueue 是基于链表的实现,适合做无界队列,当任务提交非常频繁的时候, LinkedBlockingQueue 可能迅速膨胀.
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
newCachedThreadPool 该线程池的corePoolSize 是0, maximumPoolSize 为无穷大,意味着当没有任务提交时,线程池里面没有线程,当有任务提交时,会检查是否有空闲线程,如果有的话就用空闲线程,如果没有空闲线程,就会将任务加入 SynchronousQueue 队列,
SynchronousQueue 是一种直接提交的队列,它没有容量,如果想要插入一个新的元素,就必须删除一个元素,所以提交给他的任务不会真实的保存,而是直接扔给线程池执行,当任务执行完毕后,由于corePoolSize = 0,所以空闲线程又会在指定的时间内被回收.所以如果任务太多而当任务执行不够快时,他会开启等量的线程,这样做线程池可能会开启大量的线程(后果自己想)
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
newSingleThreadExecutor 只是简单的将线程池线程数量设置为1,其余的和newFixedThreaPool 没啥区别!
