JUC-07_线程池
线程池
FutureTask
获取Runnable或callable接口的执行结果
-
实现RunnableFuture接口，该接口基础了Runnable和Future俩接口
-
也提供一个方法（适配器模式）可以让Runnable实现类执行完成后返回特定的返回值
-
配合线程池使用，一般用来异步调度获取线程执行结果
-
也可配合单个线程Thread使用
run() isDone()都类似，cas操作修改state值，run调用你Runnable或Callable的run方法
重点awaitDone()
cas+volatile，维护state，为运行线程设置多个状态，根据状态条件判断线程是执行中还是执行完成还是被中断了等等等等
常用方法
| 方法名 | 描述 |
|---|---|
| cancel() | 取消一个任务，（中断），若任务已经完成或被取消，返回false |
| isCanceled() | 判断是否被取消 |
| isDone() | 判断是否执行结束（执行中发生异常或者被取消了也是ture） |
| get() | 获取返回值，没执行完等待执行完 |
| get(long timeout Timeunit unit) | 获取返回值的等待超时版本 |
state状态值：
-
new
-
completing
-
exceptional
-
normal
-
-
canceled
-
interrupting
- interrupted
使用示例：
-
Executors + Future
-
Executors + FutureTask
-
Thread + FutureTask
package com.kaguya.demo.excutors;
import java.util.concurrent.*;
public class FutureTaskDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newSingleThreadExecutor();
// Future + Executors
Task task = new Task();
Future<?> future = executorService.submit(task);
Object r = future.get();
System.out.println(r);
executorService.shutdown();
// FutureTask + Executors
executorService = Executors.newSingleThreadExecutor();
FutureTask<Integer> futureTask = new FutureTask<Integer>(task);
executorService.submit(futureTask);
Object r1 = futureTask.get();
System.out.println(r1);
executorService.shutdown();
// Thread
new Thread(futureTask,"t1").start();
if(!futureTask.isDone()) {
System.out.println("task还没结束呢");
try{
Thread.sleep(2000);
} catch (InterruptedException e) {
}
}
Object result = new Object();
try {
// 5. 调用get()方法获取任务结果,如果任务没有执行完成则阻塞等待
result = futureTask.get();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(result);
}
}
class Task implements Callable<Integer> {
@Override
public Integer call() throws Exception {
System.out.println("Thread [" + Thread.currentThread().getName() + "] is running");
int result = 0;
for(int i = 0; i < 100;++i) {
result += i;
}
Thread.sleep(3000);
return result;
}
}
ThreadPoolExecutor
Executor：接口
ThreadPoolExecutor线程池：Executor的实现类，对线程统一分配，调优监控
-
降低系统资源消耗
-
提高响应速度
-
提高线程的可管理性
使用示例：
package com.kaguya.demo.excutors;
import java.util.concurrent.*;
public class ExecutorDemo {
public static void main(String[] args) throws InterruptedException {
// 初始创建5个核心线程，分配10个工作，始终只有5个线程，剩下5个工作阻塞，等线程空闲再执行
//ExecutorService executorService = Executors.newFixedThreadPool(5);
//for (int i = 0; i < 10; i++) {
// Runnable runnable = new Work("" + i);
// executorService.submit(runnable);
//}
//executorService.shutdown();
//executorService.awaitTermination(12000, TimeUnit.SECONDS);
//System.out.println("all works done");
//
//System.out.println("---------------------");
// 自定义线程池
// 初始线程2最大线程4 队列2 空闲线程超时回收12秒 拒绝策略
// 假设现在有4个任务，先2个任务干活，再开2个达到4，此时又来4个任务，只放2个到工作队列，其他的全部走拒绝策略丢掉
// 也就是同时能正在处理maxSize数的任务，同时最多能接受max+queue数
// 调用shutdown时，核心线程也灭了
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2,4,12,TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),new RejectHandler());
// 打印线程池参数信息
soutMonitor(executor);
for (int i = 0; i < 10; i++) {
Runnable runnable = new Work("cmd" + i);
executor.execute(runnable);
}
executor.shutdown();
Thread.sleep(3000);
System.out.println("work done");
// 打印线程池参数信息
soutMonitor(executor);
}
private static void soutMonitor(ThreadPoolExecutor executor) {
System.out.println(
String.format("[monitor] [%d/%d] Active: %d, Completed: %d, Task: %d, isShutdown: %s, isTerminated: %s",
executor.getPoolSize(),
executor.getCorePoolSize(),
executor.getActiveCount(),
executor.getCompletedTaskCount(),
executor.getTaskCount(),
executor.isShutdown(),
executor.isTerminated()));
}
}
class Work implements Runnable {
private String command;
public Work(String command) {
this.command = command;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName()+" Start. Command = "+command);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+" End.");
}
@Override
public String toString() {
return command;
}
}
// 拒绝策略
class RejectHandler implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println(r.toString() + "is rejected");
}
}
[monitor] [0/2] Active: 0, Completed: 0, Task: 0, isShutdown: false, isTerminated: false
cmd6is rejected
cmd7is rejected
cmd8is rejected
cmd9is rejected
pool-1-thread-1 Start. Command = cmd0
pool-1-thread-2 Start. Command = cmd1
pool-1-thread-3 Start. Command = cmd4
pool-1-thread-4 Start. Command = cmd5
pool-1-thread-1 End.
pool-1-thread-2 End.
pool-1-thread-3 End.
pool-1-thread-2 Start. Command = cmd2
pool-1-thread-1 Start. Command = cmd3
pool-1-thread-4 End.
pool-1-thread-2 End.
pool-1-thread-1 End.
work done
[monitor] [0/2] Active: 0, Completed: 6, Task: 6, isShutdown: true, isTerminated: true
使用详解
一个线程集合workset 和一个工作队列 workqueue，
任务会先放到workqueue中，workset从workqueue中获取任务
若work都在工作，等工作完了再去workqueue取任务
没任务了，workset阻塞
队列满了，拒绝策略
参数详解
| 参数 | 解释 | 取值 |
|---|---|---|
| corePoolSize | 核心线程数 | int |
| maxPoolSize | 最大线程数 | int |
| keepAliveTime | 空闲线程存活时间（作用超出核心线程数的线程） | Long |
| timeunit | 空闲线程存活时间单位 | TimeUnit |
| BlockingQueue | 阻塞队列 | BlockingQueue实现类，常用LinkedBlockingQueue,ArrayBlockingQueue...（二者在put(), take()任务的时均需要加锁） |
| ThreadFactory | 创建线程工厂，可自定义来给线程取名字 | 默认为DefaultThreadFactory |
| RejectHandler | 拒绝策略，可自定义拒绝策略，记录日志或等待重试 | 默认：AbortPolicy，抛出异常，CallerRunsPolicy，调用方处理任务，DiscardOledestPolicy，丢弃队列最靠前的任务，执行当前任务，DiscardPolicy，直接丢弃任务 |
Executors创建的三种线程池类型
-
newFixedThreadPool
-
空闲存活时间是0，线程空闲了也不释放线程
-
无界阻塞队列，最大值是Integer.maxValue，永远不会拒绝，引发OOM
-
-
newSingleThreadPool
-
永远只有一个线程处理任务，若线程挂了就再开一个，任务是顺序执行的
-
无界阻塞队列，问题同上
-
-
newCachedThreadPool
-
初始0，最大线程数Integer.max_value，存活时间60s
-
SynchronousQueue阻塞队列
-
执行原理
一个任务提交至线程池中
-
线程池先判断当前运行的线程是否少于corePoolSize，若是，创建新的核心线程，若都在执行工作任务，进入2
-
判断BlockingQueue是否满了，若没满，把线程放入BlockingQueue，若满了进入3
-
队列已满，若再开一个线程数超过maxPoolSize，则把它交给拒绝策略执行，没有就再开线程执行
当ThreadPoolExecutor创建新线程时，通过CAS来更新线程池的状态ctl（线程池内部维护的AtomicInteger变量）.
ctl状态：
-
running
-
stop
-
shutdown
-
tidying
-
terminated
低29位表示线程池中线程数，通过高3位表示线程池的运行状态:
关闭线程池
遍历线程池所有线程，依次调用interrupt
-
shutdown
ctl线程状态设置成SHUTDOWN，中断所有空闲线程，等待正在执行任务线程执行完才结束
-
shutdownnow
线程状态设置为STOP，停止所有线程
调用这俩其中一，isShutDown : true,所有任务都关闭后： isTerminate ： true
任务的执行
-
execute()方法只接受Runnable
-
submit()方法接受Runnabl和Callable，返回Future对象
execute -> addWorker -> runworker(getTask)
线程池的工作线程通过Worker类实现，
-
addWorkder创建新线程，创建线程时获取ReentrantLock锁
-
runWorkder方法执行任务
-
execute方法内，double check线程池状态采取不同策略，放入阻塞队列或者执行等等
runworker：
-
线程启动后，unlock释放锁，设置AQSstate为0，标识运行可中断
-
Worker执行firstTask或从workQueue中取任务
-
加锁保证不被其他Thread中断
-
检查线程池状态，若状态为中断状态，则中断线程
-
beforeExecute
-
执行任务的run方法
-
afterExecute
-
解锁
-
-
通过getTask方法从阻塞队列中获取等待的任务，如果队列中没有任务，getTask方法会被阻塞并挂起，不会占用cpu资源；
更详细源码层面这里略过，有空再补充吧（哈哈==）
任务的提交
-
等待线程池execute
-
执行FutureTask的get方法时，会把主线程封装成waitNode节点保存在waiters链表中，阻塞等待结果
-
执行完成后，cas设置waiters为null，调用LockSupport.unpark唤醒主线程
为什么不推荐用Executors
-
自定义线程池更加灵活，对于规则更加清晰（尽量用有界阻塞队列）
-
newFiexdPool和newSinglePool：主要是无界阻塞队列会引发oom
-
newCachedPool和newScheduledThreadPool：最大线程数量为Integer.max_value，过多的线程数导致oom
创建线程池需要考虑的点
-
任务的优先级
-
任务的执行时间长短
-
任务的性质（IO密集还是cpu密集）
-
任务的依赖关系
性质不同采用不同的线程池
-
cpu密集型：尽可能少的线程，N cpu + 1
-
io密集型：尽可能多的线程，N cpu *2 如数据库连接池
-
混合型：执行时间差别较小可以混一起使用，差别较大有必要拆分
-cpu密集：计算，赋值，数据处理
-io密集：如文件io，设备网络连接等
ScheduleThreadPool
-
继承自ThreadPoolExecutor
-
实现Delayed接口
-
支持定时执行任务
-
最大线程数Integer.max_value
-
SpringTask @Scheduled接口底层就是它
-
项目中定时任务一般使用集成的中间件工具，更加强大，如quartz，xxl-job等