学习JAVA线程池
一、线程池作用
线程池作用就是限制系统中执行线程的数量。
根据系统的环境情况,可以自动或手动设置线程数量,达到运行的最佳效果;少了浪费了系统资源,多了造成系统拥挤效率不高。用线程池控制线程数量,其他线程排队等候。一个任务执行完毕,再从队列的中取最前面的任务开始执行。若队列中没有等待进程,线程池的这一资源处于等待。当一个新任务需要运行时,如果线程池中有等待的工作线程,就可以开始运行了;否则进入等待队列。
基于生产者-消费者模式,其提交任务的线程相当于生产者,执行任务的线程相当于消费者,并用Runnable来表示任务
二、重要接口和类
1.接口Executor
定义了一个接收Runnable对象的方法execute
2.接口ExecutorService
是一个比Executor使用更广泛的子类接口,其提供了生命周期(运行、关闭、已终止)管理的方法,以及可跟踪一个或多个异步任务执行状况返回Future的方法。Future 表示一个任务的生命周期,并且提供了相应的方法来判断是否已经完成或取消,以及获取任务的结果和取消任务等。
创建一个什么样的ExecutorService的实例(即线程池)需要根据具体应用场景而定,不过Java给我们提供了一个Executors工厂类,它可以帮助我们很方便的创建各种类型ExecutorService线程池。
2.1 ExecutorService的使用
ExecutorService executorService = Executors.newFixedThreadPool(10);
executorService.execute(new Runnable() {
public void run() {
System.out.println("Asynchronous task");
}
});
executorService.shutdown();
2.2 ExecutorService的执行
ExecutorService有如下几个执行方法:
- execute(Runnable)
- submit(Runnable)
- submit(Callable)
- invokeAny(...)
- invokeAll(...)
2.2.1 execute(Runnable)
这个方法接收一个Runnable实例,并且异步的执行,请看下面的实例:
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.execute(new Runnable() {
public void run() {
System.out.println("Asynchronous task");
}
});
executorService.shutdown();
这个方法有个问题,就是没有办法获知task的执行结果。如果我们想获得task的执行结果,我们可以传入一个Callable的实例(下面会介绍)。
2.2.2 submit(Runnable)
submit(Runnable)和execute(Runnable)区别是前者可以返回一个Future对象,通过返回的Future对象,我们可以检查提交的任务是否执行完毕,请看下面执行的例子:
Future future = executorService.submit(new Runnable() {
public void run() {
System.out.println("Asynchronous task");
}
});
future.get(); //returns null if the task has finished correctly.
如果任务执行完成,future.get()方法会返回一个null。注意,future.get()方法会产生阻塞。
2.2.3 submit(Callable)
submit(Callable)和submit(Runnable)类似,也会返回一个Future对象,但是除此之外,submit(Callable)接收的是一个Callable的实现,Callable接口中的call()方法有一个返回值,可以返回任务的执行结果,而Runnable接口中的run()方法是void的,没有返回值。请看下面实例:
Future future = executorService.submit(new Callable(){
public Object call() throws Exception {
System.out.println("Asynchronous Callable");
return "Callable Result";
}
});
System.out.println("future.get() = " + future.get());
如果任务执行完成,future.get()方法会返回Callable任务的执行结果。注意,future.get()方法会产生阻塞,直到call方法结束返回结果。
2.2.4 invokeAny(…)
invokeAny(…)方法接收的是一个Callable的集合,执行这个方法不会返回Future,但是会返回所有Callable任务中其中一个任务的执行结果。这个方法也无法保证返回的是哪个任务的执行结果,反正是其中的某一个。请看下面实例:
ExecutorService executorService = Executors.newSingleThreadExecutor();
Set<Callable<String>> callables = new HashSet<Callable<String>>();
callables.add(new Callable<String>() {
public String call() throws Exception {
return "Task 1";
}
});
callables.add(new Callable<String>() {
public String call() throws Exception {
return "Task 2";
}
});
callables.add(new Callable<String>() {
public String call() throws Exception {
return "Task 3";
}
});
String result = executorService.invokeAny(callables);
System.out.println("result = " + result);
executorService.shutdown();
大家可以尝试执行上面代码,每次执行都会返回一个结果,并且返回的结果是变化的,可能会返回“Task2”也可是“Task1”或者其它。
2.2.5 invokeAll(…)
invokeAll(…)与 invokeAny(…)类似也是接收一个Callable集合,但是前者执行之后会返回一个Future的List,其中对应着每个Callable任务执行后的Future对象。情况下面这个实例:
ExecutorService executorService = Executors.newSingleThreadExecutor();
Set<Callable<String>> callables = new HashSet<Callable<String>>();
callables.add(new Callable<String>() {
public String call() throws Exception {
return "Task 1";
}
});
callables.add(new Callable<String>() {
public String call() throws Exception {
return "Task 2";
}
});
callables.add(new Callable<String>() {
public String call() throws Exception {
return "Task 3";
}
});
List<Future<String>> futures = executorService.invokeAll(callables);
for(Future<String> future : futures){
System.out.println("future.get = " + future.get());
}
executorService.shutdown();
2.3 ExecutorService的关闭
当我们使用完成ExecutorService之后应该关闭它,否则它里面的线程会一直处于运行状态。
如果要关闭ExecutorService中执行的线程,我们可以调用ExecutorService.shutdown()方法。在调用shutdown()方法之后,ExecutorService不会立即关闭,但是它不再接收新的任务,直到当前所有线程执行完成才会关闭,所有在shutdown()执行之前提交的任务都会被执行。
如果我们想立即关闭ExecutorService,我们可以调用ExecutorService.shutdownNow()方法。这个动作将跳过所有正在执行的任务和被提交还没有执行的任务。但是它并不对正在执行的任务做任何保证,有可能它们都会停止,也有可能执行完成。
3.接口ScheduledExecutorService
3.1 scheduleAtFixedRate() 按指定频率周期执行某个任务
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
command:执行线程
initialDelay:初始化延时
period:两次开始执行最小间隔时间
unit:计时单位
实例:
private static ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
static {
service.scheduleAtFixedRate(new Runnable() {
public void run() {
System.out.println("延迟10s开始执行,每隔100s重新执行一次任务");
}
}, 10, 100, TimeUnit.SECONDS);
}
3.2 scheduleWithFixedDelay() 按指定频率间隔执行某个任务
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
command:执行线程
initialDelay:初始化延时
period:前一次执行结束到下一次执行开始的间隔时间(间隔执行延迟时间)
unit:计时单位
实例:
private static ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
static {
service.scheduleWithFixedDelay(new Runnable() {
public void run() {
System.out.println("10秒后开始执行,本次执行结束后延迟100s开始下次执行。");
}
}, 10, 100, TimeUnit.SECONDS);
}
3.3 周期定时执行某个任务
/**
* 每天晚上8点执行一次
* 每天定时安排任务进行执行
*/
public static void executeEightAtNightPerDay() {
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
long oneDay = 24 * 60 * 60 * 1000;
long initDelay = getTimeMillis("20:00:00") - System.currentTimeMillis();
initDelay = initDelay > 0 ? initDelay : oneDay + initDelay;
executor.scheduleAtFixedRate(new Runnable() {
public void run() {
System.out.println("do something");
},
initDelay,
oneDay,
TimeUnit.MILLISECONDS);
}
/**
* 获取指定时间对应的毫秒数
* @param time "HH:mm:ss"
* @return
*/
private static long getTimeMillis(String time) {
try {
DateFormat dateFormat = new SimpleDateFormat("yy-MM-dd HH:mm:ss");
DateFormat dayFormat = new SimpleDateFormat("yy-MM-dd");
Date curDate = dateFormat.parse(dayFormat.format(new Date()) + " " + time);
return curDate.getTime();
} catch (ParseException e) {
e.printStackTrace();
}
return 0;
}
4.类Executors
通过 Executors 中的静态工厂方法来创建线程池。
4.1 newFixedThreadPool()
public static ExecutorService newFixedThreadPool(int nThreads)
创建一个定长线程池,定长线程池的大小最好根据系统资源进行设置,可控制线程最大并发数,超出的线程会在队列中等待。
每次提交一个任务就创建一个线程,直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变,如果某个线程因为执行异常而结束,那么线程池会补充一个新线程。
newFixedThreadPool内部有个任务队列,假设线程池里有3个线程,提交了5个任务,那么后两个任务就放在任务队列了,即使前3个任务sleep或者堵塞了,也不会执行后两个任务,除非前三个任务有执行完的
实例:
public class Test {
public static void main(String[] args) throws IOException, InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(2);
for (int i = 0; i < 6; i++) {
final int index = i;
System.out.println("task: " + (i + 1));
Runnable run = new Runnable() {
@Override
public void run() {
System.out.println("thread start" + index);
try {
Thread.sleep(Long.MAX_VALUE);//执行很久。。。
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("thread end" + index);
}
};
service.execute(run);
}
service.shutdown();
}
}
输出:
task: 1
task: 2
thread start0
thread start1
task: 3
task: 4
task: 5
task: 6
6个任务,只有二个任务在执行,4个在任务队列,等候执行。
4.2 newCachedThreadPool()
创建一个可缓存线程池,如果线程池中的线程在60秒未被使用就将被移除,在执行新的任务时,当线程池中有之前创建的可用线程就重用可用线程,否则就新建一条线程
public class Test {
public static void main(String[] args) {
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int index = i;
try {
Thread.sleep(index * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
cachedThreadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(index);
}
});
}
}
}
线程池为无限大,当执行第二个任务时第一个任务已经完成,会复用执行第一个任务的线程,而不用每次新建线程。
对于执行很多短期异步任务的程序而言,可提高程序性能。
4.3 newScheduledThreadPool()
创建一个定长线程池,支持定时及周期性任务执行。实例同【3.接口ScheduledExecutorService】
4.4 newSingleThreadExecutor()
创建一个单线程化的线程池,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
public class Test {
public static void main(String[] args) {
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
final int index = i;
singleThreadExecutor.execute(new Runnable() {
@Override
public void run() {
try {
System.out.println(index);
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
}
结果依次输出,相当于顺序执行各个任务。
5.类ThreadPoolExecutor
private static ExecutorService exec = new ThreadPoolExecutor(8, 8, 0L,
TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(100000),
new ThreadPoolExecutor.CallerRunsPolicy());
5.1 常用构造方法
ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler)
corePoolSize | 线程(核心)池维护线程的最少数量 |
maximumPoolSize | 线程(最大)池维护线程的最大数量 |
keepAliveTime | 线程池维护线程所允许的空闲时间 |
unit | 线程池维护线程所允许的空闲时间的单位 |
workQueue | 线程池所使用的缓冲队列 |
handler | 线程池对拒绝任务的处理策略 |
5.1.1
一个任务通过 execute(Runnable)方法被添加到线程池,任务就是一个 Runnable类型的对象,任务的执行方法就是Runnable类型对象的run()方法。当一个任务通过execute(Runnable)方法欲添加到线程池时:
l 如果此时线程池中的数量小于corePoolSize,即使线程池中的线程都处于空闲状态,也要创建新的线程来处理被添加的任务。
l 如果此时线程池中的数量等于 corePoolSize,但是缓冲队列 workQueue未满,那么任务被放入缓冲队列。
l 如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量小于maximumPoolSize,建新的线程来处理被添加的任务。
l 如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量等于maximumPoolSize,那么通过 handler所指定的策略来处理此任务。也就是:处理任务的优先级为:核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize,如果三者都满了,使用handler处理被拒绝的任务。
l 当线程池中的线程数量大于 corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止。这样,线程池可以动态的调整池中的线程数。
5.1.2 handler有四个选择
默认new ThreadPoolExecutor.AbortPolicy() | 处理程序遭到拒绝将抛出运行时RejectedExecutionException |
new ThreadPoolExecutor.CallerRunsPolicy() | 当抛出RejectedExecutionException异常时,会调用rejectedExecution方法 |
new ThreadPoolExecutor.DiscardOldestPolicy() | 抛弃旧的任务,如果执行程序尚未关闭,则位于工作队列头部的任务将被删除,然后重试执行程序(如果再次失败,则重复此过程) |
new ThreadPoolExecutor.DiscardPolicy() | 抛弃当前的任务,不能执行的任务将被删除 |
实例:(实例原文http://blog.csdn.net/pfnie/article/details/52755769)
写一个task:
public class Task implements Runnable {
protected String name;
public Task(String name) {
super();
this.name = name;
}
@Override
public void run() {
try {
System.out.println(this.name + " is running.");
Thread.sleep(500);
} catch (Exception e) {
}
}
}
1) AbortPolicy 示例
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class AbortPolicyDemo {
public static void main(String[] args) {
// 创建线程池。线程池的"最大池大小"和"核心池大小"都为1,"线程池"的阻塞队列容量为1。
ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 1, 0,
TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1));
// 设置线程池的拒绝策略为AbortPolicy
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
// 设置线程池的拒绝策略为CallerRunsPolicy
// pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 设置线程池的拒绝策略为DiscardPolicy
// pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
// 设置线程池的拒绝策略为DiscardOldestPolicy
// pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
try {
// 新建10个任务,并将它们添加到线程池中
for (int i = 0; i < 10; i++) {
Runnable myTask = new Task("task-"+i);
pool.execute(myTask);
}
} catch (RejectedExecutionException e) {
e.printStackTrace();
// 关闭线程池
pool.shutdown();
}
}
}
某一次运行结果,直接已异常的形式抛出:
task-0 is running.
java.util.concurrent.RejectedExecutionException: Task websocket.blocktest.Task@58777255 rejected from java.util.concurrent.ThreadPoolExecutor@470ae2bf[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 0]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2048)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:821)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1372)
at websocket.blocktest.Test.main(Test.java:22)
task-1 is running.
2)CallerRunsPolicy 示例
// 设置线程池的拒绝策略为CallerRunsPolicy
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
某一次运行结果,当有新任务添加到线程池被拒绝时,线程池会将被拒绝的任务添加到”线程池正在运行的线程”中去运行:
task-2 is running.
task-0 is running.
task-3 is running.
task-1 is running.
task-5 is running.
task-4 is running.
task-7 is running.
task-6 is running.
task-9 is running.
task-8 is running.
3)DiscardPolicy 示例
// 设置线程池的拒绝策略为DiscardPolicy
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
某一次运行结果,线程池pool的”最大池大小”和”核心池大小”都为1,这意味着”线程池能同时运行的任务数量最大只能是1”。线程池pool的阻塞队列是ArrayBlockingQueue,ArrayBlockingQueue是一个有界的阻塞队列,ArrayBlockingQueue的容量为1。这也意味着线程池的阻塞队列只能有一个线程池阻塞等待。由此可知,线程池*运行了2个任务。第1个任务直接放到Worker中,通过线程去执行;第2个任务放到阻塞队列中等待。其他的任务都被丢弃了。
task-0 is running.
task-1 is running.
4)DiscardOldestPolicy 示例
// 设置线程池的拒绝策略为DiscardOldestPolicy
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
某一次运行结果,当有新任务添加到线程池被拒绝时,线程池会丢弃阻塞队列中末尾的任务,然后将被拒绝的任务添加到末尾。
task-0 is running.
task-9 is running.
5.2 ThreadPoolExecutor配置(加深理解)
5.2.1 线程的创建与销毁
1)核心池大小、最大池大小和存活时间共同管理着线程的创建与销毁。
2)核心池的大小是目标的大小;线程池的实现试图维护池的大小;即使没有任务执行,池的大小也等于核心池的大小,并直到工作队列充满前,池都不会创建更多的线程。如果当前池的大小超过了核心池的大小,线程池就会终止它。
3)最大池的大小是可同时活动的线程数的上限。
4)如果一个线程已经闲置的时间超过了存活时间,它将成为一个被回收的候选者。
5)newFixedThreadPool工厂为请求的池设置了核心池的大小和最大池的大小,而且池永远不会超时
6)newCacheThreadPool工厂将最大池的大小设置为Integer.MAX_VALUE,核心池的大小设置为0,超时设置为一分钟。这样创建了无限扩大的线程池,会在需求量减少的情况下减少线程数量。
5.2.2 管理
1)ThreadPoolExecutor允许你提供一个BlockingQueue来持有等待执行的任务。任务排队有3种基本方法:无限队列、有限队列和同步移交。
2)newFixedThreadPool和newSingleThreadExectuor默认使用的是一个无限的 LinkedBlockingQueue。如果所有的工作者线程都处于忙碌状态,任务会在队列中等候。如果任务持续快速到达,超过了它们被执行的速度,队列也会无限制地增加。稳妥的策略是使用有限队列,比如ArrayBlockingQueue或有限的LinkedBlockingQueue以及 PriorityBlockingQueue。
3)对于庞大或无限的池,可以使用SynchronousQueue,完全绕开队列,直接将任务由生产者交给工作者线程
4)可以使用PriorityBlockingQueue通过优先级安排任务
6.类ScheduledThreadPoolExecutor
平时我们在执行一个定时任务时,会采用Time和TimeTask来组合处理,但是Timer和TimerTask存在一些缺陷:
1)Timer只创建了一个线程。当你的任务执行的时间超过设置的延时时间将会产生一些问题。
2)Timer创建的线程没有处理异常,因此一旦抛出非受检异常,该线程会立即终止。
JDK 5.0以后推荐使用ScheduledThreadPoolExecutor。该类属于Executor Framework,它除了能处理异常外,还可以创建多个线程解决上面的问题。
实例:
public class Test {
public static void main(String[] args) throws InterruptedException {
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(5);
for (int i = 0; i < 5; i ++){
final int temp = i + 1;
//jdk8 lambda表达式
pool.schedule(() -> {
System.out.println("第"+temp+"个炸弹爆炸时间:" + simpleDateFormat.format(new Date()));
}, temp * 5, TimeUnit.SECONDS);
//隔2秒后开始执行任务,并且在上一次任务开始后隔一秒再执行一次;
//pool.scheduleWithFixedDelay(new MyTask(), 2, 1, TimeUnit.SECONDS);
//隔6秒后执行一次,但只会执行一次。
//pool.schedule(new MyTask(), 6, TimeUnit.SECONDS);
}
pool.shutdown();
System.out.println("end main时间:" + simpleDateFormat.format(new Date()));
}
private static class MyTask implements Runnable{
@Override
public void run() {
//捕获所有的异常,保证定时任务能够继续执行
try{
System.out.println("任务开始...");
// 业务...
System.out.println("任务结束...");
}catch (Throwable e) {
// 什么也不做
}
}
}
}
在JAVAWEB开发中,执行定时任务有一个更好的选择: Quartz
上一篇: 第八章 SportsStore: 一个真实的应用程序
下一篇: 第二章 你的第一个MVC应用程序