目录

• 一、线程池（ThreadPoolExecutor）
• • 1)构造方法
  • 2）工作方式
  • 3）线程池的状态（自定义并不是这样实现）
• 自定义线程池
• • 1.流程
  • 2.execute执行流程
• 代码
• • 线程工厂
  • 拒绝策略
  • ThreadPool线程池
  • 测试

一、线程池（ThreadPoolExecutor）

1)构造方法

public ThreadPoolExecutor(  int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)

• corePoolSize 核心线程数目 (最多保留的线程数)
• maximumPoolSize 最大线程数目
• keepAliveTime 生存时间 - 针对救急线程
• unit 时间单位 - 针对救急线程
• workQueue 阻塞队列
• threadFactory 线程工厂 - 可以为线程创建时起个好名字
• handler 拒绝策略

2）工作方式

• 线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务，阻塞等待执行任务队列的任务。
• 当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue 队列排队，直到有空闲的线程。
• 如果任务队列选择了有界队列，那么任务超过了队列大小时，会创建maximumPoolSize - corePoolSize 数目的线程来救急，称为救急线程，救急线程在keepAliveTime时间后没有任务执行就进入等待，等待任务任务队列满了才唤醒。
• 如果线程到达 maximumPoolSize 且救急线程也都在执行任务，仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它著名框架也提供了实现
  • AbortPolicy 让调用者抛出 RejectedExecutionException 异常，这是默认策略
  • CallerRunsPolicy 让调用者运行任务
  • DiscardPolicy 放弃本次任务
  • DiscardOldestPolicy 放弃队列中最早的任务，本任务和最新的任务争抢进入队列的机会
• 当高峰过去后，超过corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime 和 unit 来控制。

jdk提供的拒绝策略
当等待队列满了，并且所有线程都在使用时，采取拒绝策略

3）线程池的状态（自定义并不是这样实现）

ThreadPoolExecutor 使用 int 的高 3 位来表示线程池状态，低 29 位表示线程数量

从数字上比较，TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING
这些信息存储在一个原子变量 ctl 中，目的是将线程池状态与线程个数合二为一，这样就可以用一次 cas 原子操作
进行赋值

// c 为旧值， ctlOf 返回结果为新值
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));
// rs 为高 3 位代表线程池状态， wc 为低 29 位代表线程个数，ctl 是合并它们
private static int ctlOf(int rs, int wc) { return rs | wc; }

自定义线程池

1.流程

2.execute执行流程

1. 线程池状态已经为shutdown,不向队列添加任务，结束
2. 如果当前的线程数小于核心线程数就增加核心线程,并执行任务，执行完就等待任务队列进入的新任务
3. 如果任务队列的容量大于0,即是还可以存放任务，直接进入队列（无界队列容量2^31，所以无界的任务队列基本不会执行下面的4、5、6步）
4. 创建救急线程(救急线程阻塞keepAliveTime后进入等待区)
5. 唤醒一个救急线程执行任务
6. 执行拒绝策略

代码

线程工厂

1.线程工厂接口

@FunctionalInterface
public interface LzkThreadFactory {/***  返回创建的线程* @param task* @return*/Thread newThread(Runnable task);
}

2. 默认线程工厂
   (ThreadPool静态内部类,这部分不用复制，ThreadPool类内有实现)

//(ThreadPool静态内部类,这部分不用复制，ThreadPool类内有实现)
public static class LzkDefaultThreadFactory implements LzkThreadFactory {private AtomicInteger i=new AtomicInteger(0);@Overridepublic Thread newThread(Runnable task) {return new Thread(task,"[lzkThreadPool-"+i.incrementAndGet()+"] ");}
}

拒绝策略

1. 拒绝策略接口

public interface RejectedHandler {void rejectedExecution(Runnable r, ThreadPool executor);
}

2. 拒绝策略四种实现
   (ThreadPool静态内部类,这部分不用复制，ThreadPool类内有实现)

//(ThreadPool静态内部类,这部分不用复制，ThreadPool类内有实现)
/*** 抛异常*/
public static class AbortPolicy implements RejectedHandler {public AbortPolicy() { }@Overridepublic void rejectedExecution(Runnable r, ThreadPool executor) {throw new RejectedExecutionException(executor.toString()+"中任务队列已满,任务【" + r.toString() +"】无法执行！");}
}/*** 调用者执行线程*/
public static class CallerRunsPolicy implements RejectedHandler {public CallerRunsPolicy() { }@Overridepublic void rejectedExecution(Runnable r, ThreadPool executor) {r.run();}
}
/*** 丢弃任务队列的旧任务，将任务和新任务进行竞争*/
public static class DiscardOldestPolicy implements RejectedHandler {public DiscardOldestPolicy() { }@Overridepublic void rejectedExecution(Runnable r, ThreadPool executor) {//是否停止向线程池添加任务if(!executor.getIsShutDown().get()){//丢弃任务队列的就任务executor.getWorkQueue().poll();//添加任务和新任务进行竞争executor.execute(r);}}
}/*** 直接丢弃任务，不抛异常*/
public static class DiscardPolicy implements RejectedHandler {public DiscardPolicy() { }@Overridepublic void rejectedExecution(Runnable r, ThreadPool executor) {}
}

ThreadPool线程池

public class ThreadPool {// 核心线程数private int corePoolSize;// 最大线程数private int maximumPoolSize;// 救急线程无使用时的最长时间private long keepAliveTime;// 时间单位private TimeUnit unit;// 任务队列private BlockingQueue workQueue;// 线程工厂private LzkThreadFactory threadFactory;// 拒绝策列private RejectedHandler handler;//当前线程数,和源代码不同，这里是只记录线程数量private AtomicInteger ctl=new AtomicInteger(0);//活跃的救急线程private AtomicInteger other=new AtomicInteger(0);//线程组，只有获取锁才能操作private HashSet threadSets=new HashSet();//是否停止向任务队列添加任务private AtomicBoolean isShutDown=new AtomicBoolean(false);//救急线程等待区private ReentrantLock lock=new ReentrantLock();private Condition waitSets=lock.newCondition();//get&setpublic BlockingQueue getWorkQueue() {return workQueue;}public AtomicBoolean getIsShutDown() {return isShutDown;}//构造方法public ThreadPool(int corePoolSize, int maximumPoolSize,long keepAliveTime, TimeUnit unit,BlockingQueue workQueue) {this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.keepAliveTime = keepAliveTime;this.unit = unit;this.workQueue = workQueue;//将线程工厂、拒绝策略选择默认的构造方法this.threadFactory =new LzkDefaultThreadFactory();this.handler = new AbortPolicy();}public ThreadPool(int corePoolSize, int maximumPoolSize,long keepAliveTime, TimeUnit unit,BlockingQueue workQueue,RejectedHandler handler) {this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.keepAliveTime = keepAliveTime;this.unit = unit;this.workQueue = workQueue;this.threadFactory = new LzkDefaultThreadFactory();this.handler = handler;}public ThreadPool(int corePoolSize, int maximumPoolSize,long keepAliveTime, TimeUnit unit,BlockingQueue workQueue,LzkThreadFactory threadFactory) {this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.keepAliveTime = keepAliveTime;this.unit = unit;this.workQueue = workQueue;this.threadFactory = new LzkDefaultThreadFactory();this.handler = new AbortPolicy();}public ThreadPool(int corePoolSize, int maximumPoolSize,long keepAliveTime, TimeUnit unit,BlockingQueue workQueue,LzkThreadFactory threadFactory,RejectedHandler handler) {this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.keepAliveTime = keepAliveTime;this.unit = unit;this.workQueue = workQueue;this.threadFactory = threadFactory;this.handler = handler;}//-------------------任务对象-------------------/*** 线程的任务，先执行第一个任务，之后一直尝试获取任务队列的任务执行*/class Worker implements Runnable {//创建的线程执行的第一个任务private final Runnable firstTask;private long timeOut;private TimeUnit timeUnit;public Worker(Runnable firstTask) {this.firstTask = firstTask;this.timeOut = -1L;}public Worker(Runnable firstTask, long timeOut, TimeUnit timeUnit) {this.firstTask = firstTask;this.timeOut = timeOut;this.timeUnit = timeUnit;}@Overridepublic void run(){//第一个任务Runnable task=firstTask;//阻塞获取任务队列的任务while(true){try{//是否被打断if (Thread.currentThread().isInterrupted()){ctl.decrementAndGet();threadSets.remove(this);return;}if(!Thread.currentThread().isInterrupted()&&(task!=null||(task=timeOut>=0?workQueue.poll(timeOut,timeUnit):workQueue.take())!=null)){task.run();}else {if(isShutDown.get()){//线程池已结束，结束当前救急线程return;}lock.lock();//救急线程阻塞timeOut时间获取不到任务就进入等待区try{waitSets.await();}finally {lock.unlock();}other.decrementAndGet();}//如果线程池状态为结束并且无待执行任务，则结束线程if (isShutDown.get()&&workQueue.size()==0){ctl.decrementAndGet();threadSets.remove(this);return;}}catch (InterruptedException ie) {//非法打断，重新打断Thread.currentThread().interrupt();}catch (Exception e){e.printStackTrace();}finally {task=null;}}}}/*** 提交任务* @param task*/public void execute(Runnable task){//若为空不执行if(task==null){return;}if(isShutDown.get()){//线程池状态已经为shutdown,不向队列添加任务return;}else if(ctl.get()//1、如果当前的线程数小于核心线程数就增加核心线程,并执行任务，执行完就等待任务队列进入的新任务Thread thread = threadFactory.newThread(new Worker(task));synchronized (threadSets){threadSets.add(thread);}thread.start();ctl.incrementAndGet();}else if(workQueue.remainingCapacity()>0){//2、如果任务队列的容量大于0,即是还可以存放任务，无需拒绝,直接进入队列（无界队列容量2^31）workQueue.add(task);}else if(workQueue.remainingCapacity()==0&&ctl.get()//3、创建救急线程(救急线程阻塞keepAliveTime后进入等待区)Thread thread = threadFactory.newThread(new Worker(task,keepAliveTime,unit));synchronized (threadSets){threadSets.add(thread);}thread.start();//记录活跃救急线程和总线程数other.incrementAndGet();ctl.incrementAndGet();}else if(workQueue.remainingCapacity()==0&&ctl.get()==maximumPoolSize&&other.get()//4、唤醒一个救急线程执行任务waitSets.signal();other.incrementAndGet();}else if(workQueue.remainingCapacity()==0&&ctl.get()==maximumPoolSize&&other.get()==maximumPoolSize-corePoolSize){//5、执行拒绝策略handler.rejectedExecution(task,this);}}/*** 停止添加任务，停止线程池*/public void shutdown(){//标志状态this.isShutDown.compareAndSet(false,true);//等待任务被分配到每个线程while(workQueue.size()>0){}//循环判断是否所以线程都结束了while(ctl.get()>0){synchronized (threadSets){for (Thread t:threadSets){if("TERMINATED".equals(t.getState())){ctl.decrementAndGet();threadSets.remove(t);continue;}//waiting:在阻塞获取任务if("WAITING".equals(t.getState().toString())){ctl.decrementAndGet();//打断在阻塞的线程t.interrupt();}}}}}/*** 现在停止线程池*/public void shutdownNow(){//标志状态this.isShutDown.compareAndSet(false,true);//循环打断线程while(ctl.get()>0){synchronized (threadSets){for (Thread t:threadSets){ctl.decrementAndGet();//打断在阻塞的线程t.interrupt();}}}}//默认的线程工厂public static class LzkDefaultThreadFactory implements LzkThreadFactory {private AtomicInteger i=new AtomicInteger(0);@Overridepublic Thread newThread(Runnable task) {return new Thread(task,"[lzkThreadPool-"+i.incrementAndGet()+"] ");}}//----------------------------拒绝策略------------------------------/*** 抛异常*/public static class AbortPolicy implements RejectedHandler {public AbortPolicy() { }@Overridepublic void rejectedExecution(Runnable r, ThreadPool executor) {throw new RejectedExecutionException(executor.toString()+"中任务队列已满,任务【" + r.toString() +"】无法执行！");}}/*** 调用者执行线程*/public static class CallerRunsPolicy implements RejectedHandler {public CallerRunsPolicy() { }@Overridepublic void rejectedExecution(Runnable r, ThreadPool executor) {r.run();}}/*** 丢弃任务队列的旧任务，将任务和新任务进行竞争*/public static class DiscardOldestPolicy implements RejectedHandler {public DiscardOldestPolicy() { }@Overridepublic void rejectedExecution(Runnable r, ThreadPool executor) {//是否停止向线程池添加任务if(!executor.getIsShutDown().get()){//丢弃任务队列的就任务executor.getWorkQueue().poll();//添加任务和新任务进行竞争executor.execute(r);}}}/*** 直接丢弃任务，不抛异常*/public static class DiscardPolicy implements RejectedHandler {public DiscardPolicy() { }@Overridepublic void rejectedExecution(Runnable r, ThreadPool executor) {}}
}

测试

public class Demo {public static void main(String[] args) throws InterruptedException {ThreadPool threadPool=new ThreadPool(2, 4, 3, TimeUnit.SECONDS,new LinkedBlockingQueue<>(10), new LzkThreadFactory() {AtomicInteger tid=new AtomicInteger(0);@Overridepublic Thread newThread(Runnable task) {return new Thread(task,"t"+tid.incrementAndGet());}},new ThreadPool.DiscardPolicy());AtomicInteger atomicInteger=new AtomicInteger(0);CountDownLatch countDownLatch=new CountDownLatch(10);for (int j=0;j<14;j++){threadPool.execute(()->{for (int i=0;i<100;i++){atomicInteger.incrementAndGet();}System.out.println(Thread.currentThread().getName()+" "+atomicInteger.get());countDownLatch.countDown();});}try {countDownLatch.await();for (int j=0;j<4;j++){threadPool.execute(()->{for (int i=0;i<100;i++){atomicInteger.incrementAndGet();}System.out.println(Thread.currentThread().getName()+" "+atomicInteger.get());});}} catch (InterruptedException e) {e.printStackTrace();}threadPool.shutdown();System.out.println("结束");}
}