享元模式

属于23种设计模式种的一种享元模式，具体定义看这篇博客

具体体现包装类
在JDK中 Boolean，Byte，Short，Integer，Long，Character 等包装类提供了 valueOf 方法，例如 Long 的valueOf 会缓存 -128~127 之间的 Long 对象，在这个范围之间会重用对象，大于这个范围，才会新建 Long 对象

public static Long valueOf(long l) {final int offset = 128;if (l >= -128 && l <= 127) { // will cachereturn LongCache.cache[(int)l + offset];}return new Long(l);
}

注意：
Byte, Short, Long 缓存的范围都是 -128~127
Character 缓存的范围是 0~127
Integer的默认范围是 -128~127

• 最小值不能变
• 但最大值可以通过调整虚拟机参数 -Djava.lang.Integer.IntegerCache.high 来改变

Boolean 缓存了 TRUE 和 FALSE

手写数据库连接池

例如：一个线上商城应用，QPS 达到数千，如果每次都重新创建和关闭数据库连接，性能会受到极大影响。 这时预先创建好一批连接，放入连接池。一次请求到达后，从连接池获取连接，使用完毕后再还回连接池，这样既节约了连接的创建和关闭时间，也实现了连接的重用，不至于让庞大的连接数压垮数据库

为什么需要线程池

• 虽然创建和销毁线程比较小（和进程相比），但是当系统中线程数量比较多，这个开销就比较可观的，所以引入线程池

• 目的就是让某些对象被多次重复利用，减少频繁创建和销毁对象带来的开销问题（这些对象一定可以复用的）

• 类比数据库池，创建和销毁数据库的连接就是一个比较耗时的操作，表示当前用户不再使用此连接，就回收到连接池中（同一个连接可以被多个用户使用多次，减少了每次创建和销毁连接的系统开销）

• 同样的，不同的线程也就是run方法的内容不同，线程的大致流程都是一样的，因此为了避免重复创建和销毁线程带来的开销，可以让线程"复用"起来(线程池最大的好处就是每次减少启动和销毁线程的损耗，提供高时间和空间的利用率)线程池内部创建好了若干个线程，这些线程都是runnable，只需要从系统中取出任务，就可以立即开始执行

线程池主要有以下三个原因：

• 创建/销毁线程需要消耗系统资源，线程池可以复用已创建的线程。
• 控制并发的数量。并发数量过多，可能会导致资源消耗过多，从而造成服务器崩溃。（主要原因）
• 可以对线程做统⼀管理。

自定义线程池

• 我们的线程池中的线程就是对应的消费者，主线程是生产者，消费者从阻塞队列拿任务，主线程填入任务

自定义拒绝策略接口

@FunctionalInterface // 拒绝策略
interface RejectPolicy<T> {void reject(BlockedQueue<T> queue, T task);
}

• 什么是拒绝策略，如果当任务很多的时候，我们线程池处理不了，也就是我们的阻塞队列满了，这时候就需要对应的处理方法，这里的方法就是拒绝策略
• 这里为什么要定义一个接口，我们知道不同的场景下需要不同的策略，如果我们写死在我们的阻塞队列，那么不符合开闭原则，也会有大量的if-else
• 所以采用了策略模式——这里的接口就是抽象策略类 在具体的条件下实现具体的策略

自定义任务队列

public class BlockedQueue<T> {//1任务队列private Deque<T> queue = new ArrayDeque<>();//2 锁private ReentrantLock lock = new ReentrantLock();//3 生产者条件变量private Condition fullWaitSet=lock.newCondition();//4消费者条件变量private Condition emptyWaitSet=lock.newCondition();//5容量private int capcity;public BlockedQueue(int capcity) {this.capcity = capcity;}//阻塞获取public T take(){lock.lock();try {while (queue.isEmpty()){try {emptyWaitSet.await();} catch (InterruptedException e) {e.printStackTrace();}}T t = queue.removeFirst();fullWaitSet.signal();return t;}finally {lock.unlock();}}//带时间的阻塞获取public T poll(long timeOut, TimeUnit unit){lock.lock();try {long nanos = unit.toNanos(timeOut);while (queue.isEmpty()){try {if(nanos<=0){return null;}nanos = emptyWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {e.printStackTrace();}}T t = queue.removeFirst();fullWaitSet.signal();return t;}finally {lock.unlock();}}//阻塞添加public void put(T task){lock.lock();try {while (queue.size() == capcity){try {System.out.println("等待加入任务队列"+task);fullWaitSet.await();} catch (InterruptedException e) {e.printStackTrace();}}System.out.println("加入了任务队列"+task);queue.addLast(task);emptyWaitSet.signal();}finally {lock.unlock();}}//带超时时间阻塞添加public boolean offer(T task, long timeOut, TimeUnit timeUnit){lock.lock();try {long nanos = timeUnit.toNanos(timeOut);while (queue.size()== capcity){if (nanos<=0){return false;}System.out.println("等待加入任务队列"+task);try {nanos = fullWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {e.printStackTrace();}}System.out.println("加入了任务队列"+task);queue.addLast(task);emptyWaitSet.signal();return true;}finally {lock.unlock();}}public int size() {lock.lock();try {return queue.size();} finally {lock.unlock();}}public void tryPut(RejectPolicy<T> rejectPolicy,T task){lock.lock();try {if(queue.size() == capcity){rejectPolicy.reject(this,task);}else {queue.addLast(task);emptyWaitSet.signal();}}finally {lock.unlock();}}
}

• 因为ArrayDeque不是线程安全的，所以需要加锁

自定义线程池

public class ThreadPool {//任务阻塞队列private BlockedQueue<Runnable> taskQueue;//线程集合 作为线程池的容器//这里的Worker使用了适配器模式private HashSet<Worker> workers = new HashSet<>();//核心线程池private int coreSize;//获取任务时的超时时间private long timeOut;//进行时间转换private TimeUnit timeUnit;//具体的策略实现private RejectPolicy<Runnable> rejectPolicy;public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity,RejectPolicy<Runnable> rejectPolicy) {this.coreSize = coreSize;this.timeOut = timeout;this.timeUnit = timeUnit;this.taskQueue = new BlockedQueue<>(queueCapcity);this.rejectPolicy = rejectPolicy;}public void execute(Runnable task){// 当任务数没有超过 coreSize 时，直接交给 worker 对象执行// 如果任务数超过 coreSize 时，加入任务队列暂存synchronized (workers){if(workers.size() < coreSize){Worker worker = new Worker(task);workers.add(worker);worker.start();}else {//1死等//2带超时等待//3让调用这放弃任务执行//4让调用者抛出异常//5让调用者自己执行任务taskQueue.tryPut(rejectPolicy,task);}}}class Worker extends Thread{private Runnable task;public Worker(Runnable task){this.task=task;}@Overridepublic void run() {//执行任务//1当task不为空 执行任务//2当task执行完毕，再接着从任务队列获取任务并执行while (task != null || (task = taskQueue.poll(timeOut,timeUnit))!= null){try {task.run();}catch (Exception e){e.printStackTrace();}finally {task=null;}}}}
}

测试

public class PoolTest {public static void main(String[] args) {ThreadPool threadPool = new ThreadPool(1,1000, TimeUnit.MILLISECONDS,1,(queue, task) ->{//1死等queue.put(task);//2超时等待
//                    queue.poll(1500,TimeUnit.MILLISECONDS);// 3) 让调用者放弃任务执行// log.debug("放弃{}", task);// 4) 让调用者抛出异常// throw new RuntimeException("任务执行失败 " + task);// 5) 让调用者自己执行任务
//                    task.run();});for (int i = 0; i < 4; i++) {int j = i;threadPool.execute(() -> {try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(j);});}}
}

JDK中的线程池

常用的线程池的类和接口的之间的关系

线程池状态

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

状态名	高 3 位	接收新任 务	处理阻塞队列任 务	说明
RUNNING	111	Y	Y
SHUTDOWN	000	N	Y	不会接收新任务，但会处理阻塞队列剩余 任务
STOP	001	N	N	会中断正在执行的任务，并抛弃阻塞队列 任务
TIDYING	010	-	-	任务全执行完毕，活动线程为 0 即将进入 终结
TERMINATED	011	-	-	终结状态

从数字上比较，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; }

构造方法

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

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

线程池的工作流程

• 线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。
  • 当线程池中线程数小于核心线程数，就都会开一个新线程来处理任务
• 当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue 队列排队，直到有空闲的线程。
• 如果队列选择了有界队列，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线程来救急。
• 如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它著名框架也提供了实现
  • AbortPolicy 让调用者抛出 RejectedExecutionException 异常，这是默认策略
  • CallerRunsPolicy 让调用者运行任务
  • DiscardPolicy 放弃本次任务
  • DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之
• Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方便定位问题
• Netty 的实现，是创建一个新线程来执行任务
• ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略
• PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略当高峰过去后，超过corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime 和 unit 来控制

拒绝策略

• CallerRunsPolicy 返回给线程的调用者处理
• AbortPolicy 超出负荷的任务直接拒绝，抛出异常
• DiscardOldestPolicy：丢弃队列中最老的任务（排队直接最长的任务）
• DisCardPolicy：丢弃新来的任务

Execuctors

这是一个工具类，实现四大线程池

在 Java 中 Executor 和 Executors 的区别

• Executors 工具类的不同方法按照我们的需求创建了不同的线程池，来满足业务的需求。Executor 接口对象能执行我们的线程任务。
• ExecutorService 接口继承了 Executor 接口并进行了扩展，提供了更多的方法我们能获得任务执行的状态并且可以获取任务的返回值。使用 ThreadPoolExecutor 可以创建自定义线程池

newFixedThreadPool

  public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());}

特点——固定大小线程数

• 核心线程数 == 最大线程数（没有救急线程被创建），因此也无需超时时间
• 阻塞队列是无界的，可以放任意数量的任务

评价 适用于任务量已知，相对耗时的任务

newCachedThreadPool

public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());}

特点——动态变化的线程池

• 核心线程数是 0， 最大线程数是 Integer.MAX_VALUE，救急线程的空闲生存时间是 60s
  • 意味着全部都是救急线程（60s 后可以回收）
  • 救急线程可以无限创建
• 队列采用了 SynchronousQueue 实现特点是，它没有容量，没有线程来取是放不进去的（一手交钱、一手交货）

评价 整个线程池表现为线程数会根据任务量不断增长，没有上限，当任务执行完毕，空闲 1分钟后释放线程。 适合任务数比较密集，但每个任务执行时间较短的情况

public class SynchronousQueueTest {public static void main(String[] args) {SynchronousQueue<Integer> integers = new SynchronousQueue<>();new Thread(()->{try {System.out.println("putting 1");integers.put(1);System.out.println("putted 1");System.out.println("putting 2");integers.put(2);System.out.println("putted 2");} catch (InterruptedException e) {e.printStackTrace();}},"t1").start();new Thread(()->{try {System.out.println("taking 1");integers.take();} catch (InterruptedException e) {e.printStackTrace();}},"t2").start();new Thread(()->{try {System.out.println("taking 2");integers.take();} catch (InterruptedException e) {e.printStackTrace();}},"t3").start();}
}

putting 1
taking 1
putted 1
putting 2
taking 2
putted 2

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}

使用场景：

• 希望多个任务排队执行。线程数固定为 1，任务数多于 1 时，会放入无界队列排队。任务执行完毕，这唯一的线程也不会被释放。

和普通线程的区别：

• 自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一个线程，保证池的正常工作

• Executors.newSingleThreadExecutor() 线程个数始终为1，不能修改

  • FinalizableDelegatedExecutorService 应用的是装饰器模式，只对外暴露了 ExecutorService 接口，因此不能调用 ThreadPoolExecutor 中特有的方法

• Executors.newFixedThreadPool(1) 初始时为1，

  • 以后还可以修改对外暴露的是 ThreadPoolExecutor 对象，可以强转后调用 setCorePoolSize 等方法进行修改

• 单线程有意义吗，和我们创建一个线程的区别在哪，我们创建一个线程只能执行一个任务，一个任务执行完就销毁了，但是单线程池，虽然同一个时间只能执行一个任务，当这个任务执行结束之后，继续在工作队列调度一个新的任务继续执行（还是减少了销毁和创建线程的开销）

newScheduledThreadPool

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize);}

在『任务调度线程池』功能加入之前，可以使用 java.util.Timer 来实现定时功能，Timer 的优点在于简单易用，但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务。

public class TimerTest {public static void main(String[] args) {Timer timer = new Timer();TimerTask task1= new TimerTask(){@Overridepublic void run() {try {System.out.println("task 1");TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}}};TimerTask task2=new TimerTask(){@Overridepublic void run() {try {System.out.println("task 2");TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}}};// 使用 timer 添加两个任务，希望它们都在 1s 后执行// 但由于 timer 内只有一个线程来顺序执行队列中的任务，因此『任务1』的延时，影响了『任务2』的执行timer.schedule(task1, 1000);timer.schedule(task2, 1000);}
}
20:46:09.444 c.TestTimer [main] - start...
20:46:10.447 c.TestTimer [Timer-0] - task 1
20:46:12.448 c.TestTimer [Timer-0] - task 2

使用 ScheduledExecutorService 改写：

ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
// 添加两个任务，希望它们都在 1s 后执行
executor.schedule(() -> {System.out.println("任务1，执行时间：" + new Date());try { Thread.sleep(2000); } catch (InterruptedException e) { }
}, 1000, TimeUnit.MILLISECONDS);
executor.schedule(() -> {System.out.println("任务2，执行时间：" + new Date());
}, 1000, TimeUnit.MILLISECONDS);任务1，执行时间：Thu Jan 03 12:45:17 CST 2019
任务2，执行时间：Thu Jan 03 12:45:17 CST 2019

ExecutorService 接口常用方法

提交任务

// 执行任务
void execute(Runnable command);// 提交任务 task，用返回值 Future 获得任务执行结果
<T> Future<T> submit(Callable<T> task);// 提交 tasks 中所有任务
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)throws InterruptedException;// 提交 tasks 中所有任务，带超时时间
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)throws InterruptedException;// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消
<T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消，带超时时间
<T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;

线程池中 submit() 和 execute() 方法有什么区别
相同点：

• 相同点就是都可以开启线程执行池中的任务。

不同点：

• 接收参数：execute()只能执行 Runnable 类型的任务。submit()可以执行 Runnable 和Callable 类型的任务。
• 返回值：submit()方法可以返回持有计算结果的 Future 对象，而execute()没有
• 异常处理：submit()方便Exception处理

关闭线程池

/*
线程池状态变为 SHUTDOWN
- 不会接收新任务
- 但已提交任务会执行完
- 此方法不会阻塞调用线程的执行
*/
void shutdown();

 public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// 修改线程池状态advanceRunState(SHUTDOWN);// 仅会打断空闲线程interruptIdleWorkers();onShutdown(); // 扩展点 ScheduledThreadPoolExecutor} finally {mainLock.unlock();}// 尝试终结(没有运行的线程可以立刻终结，如果还有运行的线程也不会等)tryTerminate();}

/*
线程池状态变为 STOP
- 不会接收新任务
- 会将队列中的任务返回
- 并用 interrupt 的方式中断正在执行的任务
*/
List<Runnable> shutdownNow();

 public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// 修改线程池状态advanceRunState(STOP);// 打断所有线程interruptWorkers();// 获取队列中剩余任务tasks = drainQueue();} finally {mainLock.unlock();}// 尝试终结tryTerminate();return tasks;}

其他方法

// 不在 RUNNING 状态的线程池，此方法就返回 true
boolean isShutdown();
// 线程池状态是否是 TERMINATED
boolean isTerminated();
// 调用 shutdown 后，由于调用线程并不会等待所有任务运行结束，因此如果它想在线程池 TERMINATED 后做些事情，可以利用此方法等待
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

如何正确处理执行任务异常

方法1：主动捉异常

 ExecutorService pool = Executors.newFixedThreadPool(1);pool.submit(() -> {try {log.debug("task1");int i = 1 / 0;} catch (Exception e) {log.error("error:", e);}});

方法2：使用 Future

    ExecutorService pool = Executors.newFixedThreadPool(1);Future<Boolean> f = pool.submit(() -> {log.debug("task1");int i = 1 / 0;return true;});log.debug("result:{}", f.get());

• 如果报异常，那么我们submit返回的就是我们的异常信息

如何使用线程池

阿里编码归约：尽量不要使用内置线程池，最好根据实际的业务需求，定制线程池自己new ThreadPoolExeutor对象，传递相关参数

创建多少线程池合适

• 过小会导致程序不能充分地利用系统资源、容易导致饥饿
• 过大会导致更多的线程上下文切换，占用更多内存

CPU 密集型运算

CPU密集的意思是该任务需要大量的运算，而没有阻塞，CPU一直全速运行。CPU密集任务只有在真正的多核CPU上才可能得到加速(通过多线程)，而在单核CPU上，无论你开几个模拟的多线程，该任务都不可能得到加速，因为CPU总的运算能力就那样。

• 通常采用 cpu 核数 + 1 能够实现最优的 CPU 利用率，+1 是保证当线程由于页缺失故障（操作系统）或其它原因导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费

I/O 密集型运算

IO密集型，即该任务需要大量的IO，即大量的阻塞。在单线程上运行IO密集型的任务会导致浪费大量的CPU运算能力浪费在等待。所以在IO密集型任务中使用多线程可以大大的加速程序运行，即时在单核CPU上，这种加速主要就是利用了被浪费掉的阻塞时间。

• IO密集型时，大部分线程都阻塞，故需要多配置线程数，2*cpu核数

经验公式如下

线程数 = 核数 * 期望 CPU 利用率 * 总时间(CPU计算时间+等待时间) / CPU 计算时间

• 例如 4 核 CPU 计算时间是 50% ，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式
  • 4 * 100% * 100% / 50% = 8
• 例如 4 核 CPU 计算时间是 10% ，其它等待时间是 90%，期望 cpu 被 100% 利用，套用公式
  • 4 * 100% * 100% / 10% = 40