为了更好的控制多线程,JDK提供了一套线程框架Executor,帮助开发人员有效的进行线程控制。
Executors创建线程的方法:
newFixedThreadPool方法:该方法返回一个固定数量的线程池,该方法的线程数始终不变,当有一个任务提交时,若有空闲的线程,则立即执行。若无空闲的线程再会加入到一个队列里面去。
newSingleThreadExecutor : 创建一个线程的线程池,若空闲则执行。否则放入一个队列之中。
newCachedThreadPool: 返回一个可根据实际情况调整线程个数的线程池,不限制最大的线程数量,若用空闲的线程则执行任务。若无任务则不创建线程,并且每个线程在空闲60秒后会回收。
newScheduledThreadPool方法: 返回一个ScheduledExecutorService对象,该线程池可以指定线程数。
ThreadPoolExecutor 介绍:
在了解Executors的创建方法之前,必须先了解一个类 ThreadPoolExecutor ,因为Executors都是基于这个类去实现的,我们也可以通过这个类去自定义我们需要的线程池。
这个自定义线程池常用的构造方法:
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
}
corePoolSize :核心线程数(在进队列之前,可以运行的线程数)
maximumPoolSize :当队列装满任务之后,则会立即创建线程,直到maximumPoolSize定义的上限。
keepAliveTime : 线程空闲之后,能够等待的时间。
unit: 等待的时间单位。
workQueue :corePoolSize 线程满了之后,存放任务的队列(主要分为有界和无界队列)。
handler: 拒绝策略
有如下代码:
public class UseThreadPoolExecutor1 implements Runnable {
private static AtomicInteger atomicInteger = new AtomicInteger(0);
public void run() {
try {
int count = atomicInteger.incrementAndGet();
System.out.println("任务 : " + count);
System.out.println(System.currentTimeMillis());
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws InterruptedException {
//有界队列
BlockingQueue blockingQueue = new ArrayBlockingQueue(10);
//核心线程数是1,最大线程数是1,空闲线程等待时间60秒,初始大小为10的有界队列
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, blockingQueue);
//通过线程池,创建线程
for(int i = 0 ; i < 20 ; i++){
threadPoolExecutor.execute(new UseThreadPoolExecutor1());
}
Thread.sleep(1000);
System.out.println("blockingQueue size :" + blockingQueue.size());
Thread.sleep(2000);
}}
运行结果:
我们总共创建了20个线程,但是可以看到的是总共运行了11个线程,而且线程之间每2秒执行一次,证明了线程是被一个一个创建的。其中还抛了一个异常,另外九个线程去哪里了呢?
在这里corePoolSize 是1 ,workQueue 的大小是10.当调用线程的时候会根据 corePoolSize 创建一个线程,当线程多余一个时则将任务放入队列workQueue 。但是workQueue 也只能装10个。总共20个线程,多出了九个,这个时候会去判断maximumPoolSize的大小,如果maximumPoolSize是10,则就说明超出了队列的承受范围,我们总共可以创建10个线程,这个时候就会立即创建九个线程来运行。但是我们这里maximumPoolSize只是1,所以导致剩余的9个线程无法进入队列,又无法创建,所以出现了异常。出现这样的异常,我们可以通过RejectedExecutionHandler 去进行一个相应的处理。
如下我们将代码改一下,
public class UseThreadPoolExecutor1 implements Runnable {
private static AtomicInteger atomicInteger = new AtomicInteger(0);
public void run() {
try {
int count = atomicInteger.incrementAndGet();
System.out.println("创建时间: " + System.currentTimeMillis());
System.out.println("任务 : " + count);
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws InterruptedException {
//有界队列
BlockingQueue blockingQueue = new ArrayBlockingQueue(10);
//核心线程数是1,最大线程数是10,空闲线程等待时间60秒,初始大小为10的有界队列
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 10, 60, TimeUnit.SECONDS, blockingQueue);
//通过线程池,创建线程
for (int i = 0; i < 20; i++) {
threadPoolExecutor.execute(new UseThreadPoolExecutor1());
}
Thread.sleep(1000);
System.out.println("blockingQueue size :" + blockingQueue.size());
Thread.sleep(2000);
}}
运行结果:
在这里将maximumPoolSize的值设置为10,结果就变为前10个线程基本上是同一时间创建的,后面10个线程是相隔2秒之后一同创建。当然,这里也可以将blockingQueue改为一个无界队列LinkedBlockingQueue,一样可以保证我们的任务能够全部运行。
分析Executors代码:
看了自定义线程池threadPoolExecutor之后,我们可以看看Executors创建线程池的相应代码:
newFixedThreadPool:
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
创建指定的固定线程数,当线程任务结束直接释放线程,使用的无界队列可以保证任务全部执行。
newSingleThreadExecutor:
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
同 newFixedThreadPool类似,但是内部固定只创建一个线程
newCachedThreadPool:
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
创建一个上不封顶的线程池,60空闲后释放。SynchronousQueue可以保证一开始就被take阻塞中,当有任务put时,立即被消费创建线程。
newScheduledThreadPool:
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
最终还是调用的ThreadPoolExecutor
该方法允许创建一个延迟队列的线程池。