【JavaEE -- 多线程3 - 多线程案例】

1.单例模式

单例模式：是一种设计模式，某个类，在一个进程中只创建出一个实例（对象），对代码进行一个更严格的校验和检查。

实现单例模式最基础的实现方式：

• 饿汉模式：
• 懒汉模式

1.1 饿汉模式的实现方法

单例模式中一种简单的写法，饿汉：形容创建实例非常迫切，实例是在类加载的时候就创建了，创建时机非常早，相当于程序一启动，实例就创建了。

class Singleton {// 在这个Singleton 被加载的时候，就会初始化这个静态成员private static Singleton instance = new Singleton();// instance 指向的这个对象，就是唯一的一个对象public static Singleton getInstance() {// 对于饿汉模式，getInstance直接返回Instance实例，这个操作本质上是读操作，在多线程情况下读取同一个变量是线程安全的return instance;}private Singleton() {}
}public class ThreadDemo26 {public static void main(String[] args) {// Singleton singleton = new Singleton(); //Singleton s = Singleton.getInstance();Singleton s2 = Singleton.getInstance();System.out.println(s == s2);}
}

1.2 懒汉模式的实现方法

懒汉模式：创建实例的时机比较晚，只到第一次使用的时候才会创建实例。
注意：

• 在这个引用 指向唯一实例，这个引用先初始化为null，而不是立即创建实例
• 由于在 instance = new SingletonLazy(); 实例化的时候有读有写，在多线程下是不安全的，会出现指令重排序的线程安全问题，通过添加volatile解决。
• instance = new SingletonLazy(); 拆成三大步骤：
  1. 申请一段内存空间
  2. 在这内存上调用构造方法，创建出这个实例
  3. 把这个内存地址赋值给Instance引用变量
     正常是123，但是在多线程下可能132，就会出现问题
• 如果InStance为null，就说明首次调用，首次调用就需要考虑到线程安全问题，如果非null，就说明是后续的调用，就不必加锁，即双重校验锁。

** volatile**：

1. 保证内存可见性，每次访问变量都必须重新读取内存，而不会优化到寄存器/缓存中。
2. 禁止指令重排序，针对这个volatile修饰的变量的读写操作相关指令，是不能被重排序的。

class SingletonLazy {// 这个引用指向唯一实例，这个引用先初始化为null，而不是立即创建实例// 在这里添加volatile 避免重排序引起的线程安全问题private volatile static SingletonLazy instance = null;private static Object locker = new Object();// 在懒汉模式，有读也有写 instance = new SingletonLazy();,在多线程下是不安全的，且不是单例模式了，// 1.通过加锁 synchronized  然后再 把if 和 new两个操作打包成一个原子的public static SingletonLazy getInstance() {// 2.如果Istance 为null，就说明首次调用，首次调用就需要考虑到线程安全问题// 如果非null，就说明是后续的调用，就不必加锁// 双重校验锁if (instance == null) {synchronized (locker) {if (instance == null) {instance = new SingletonLazy();// 3.还会出现指令重排序引起线程安全，通过添加volatile解决/**instance = new SingletonLazy(); 拆成三大步骤* 1.申请一段内存空间* 2.在这个内存上调用构造方法，创建出这个实例* 3.把这个内存地址赋值给Instance引用变量* 正常是123，但是在多线程下可能132 就会出现问题*/}}}return instance;}private SingletonLazy() {}
}
public class ThreadDemo27 {public static void main(String[] args) {SingletonLazy s1 = SingletonLazy.getInstance();SingletonLazy s2 = SingletonLazy.getInstance();System.out.println(s1 == s2);}
}

2. 阻塞队列

阻塞队列是一种特殊的队列，遵守先进先出的原则。
阻塞队列是一种线程安全的数据结构，包含两个特性：

• 当队列满的时候，继续入队列就会阻塞，直到有其他线程从队列中取走元素。
• 当队列空的时候，继续出队列也会阻塞，直到有其他线程往队列中插入元素。

基于阻塞队列，就可以实现 生产消费者模型（是一种多线程编程的方法）。

2.1 引入生产消费者模型的意义：

1. 解耦合，即把代码的耦合程度，从高降低，在实际开发中，经常涉及到分布式系统，服务器整个功能不是由一个服务器全部完成的，而是每个服务器负责一部分功能，通过服务器之间的网络通信，最终完成整个功能。
   如公网内 的电商网站客户端，获取到主页信息，机房内部网络中 有入口服务器A，用户服务器B，商品服务器C。在这个模型中，A代码就需要涉及到一些和B相关的操作，同样B也涉及A，A和C中的代码也相互涉及。如果B或者C挂了，对A的影响非常大，即为高耦合。
   

引入生产消费者模型之后就可降低耦合，即添加阻塞队列：
上述模型中，A和B、C都不是直接交互了，而是通过阻塞队列传话，如果B或者C挂了，对A的影响几乎没有。

2. 削峰填谷：如下图模型，当请求多了，A的请求数量会增加很多，B用户服务器（找到对应用户信息）和C商品服务器（从数据库中匹配商品）都会有很大的影响。
   
   添加阻塞队列之后：即使外界的请求出现峰值，队列没有业务逻辑，只是存储数据抗压能力很强。有效的防止了B和C被冲击挂掉。
   

2.2 阻塞队列put方法和take方法

• put(): put和offer都是入队列，而put带有阻塞功能，没带阻塞功能，队列满了会返回结果。
• take():取出元素的时候，带有阻塞功能，判定如果队列为空，就进行wait阻塞等待。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;public class ThreadDemo28 {public static void main(String[] args) throws InterruptedException {BlockingQueue<String> queue = new ArrayBlockingQueue<>(100);// put和offer都是入队列 ，而 put带有阻塞功能，没带阻塞，队列满了会返回结果queue.put("aaa");// take 也带有阻塞功能String elem = queue.take();System.out.println("elem = " + elem);elem = queue.take();System.out.println("elem = " + elem);}
}

2.3 实现阻塞队列–重点

1）先实现普通队列，基于数组来实现（环形队列），区分队列空和队列满：

1. 浪费一个格子，定义一个head队头和tail队尾，tail最多走到head的前一个位置
2. 引入size变量

2）再加上线程安全
3）再加上阻塞功能

• 队列满了，添加wait进行阻塞，队列不满，即出队列成功后进行notify唤醒
• 队列空了，再出队列，同样也需要阻塞，同样是在另一个入队列成功后的线程中唤醒

class MyBlockingQueue {private String[] elems = null;private int head = 0;private int tail = 0;private int size = 0;public MyBlockingQueue(int capacity) {elems = new String[capacity];}private Object locker = new Object();public void put(String elem) throws InterruptedException {synchronized (locker) {// 使用while：wait可能会被提前唤醒（当条件还没满足，就被唤醒了）while (size >= elems.length) {// 队列满了 实现阻塞locker.wait(); //在Java标准库推荐使用wait搭配while循环，多一/N次确认操作}// 如果加锁不包含 while判断队列是否为满，在多线程下就会导致当入队列就会多入一个// 新的元素要放到 tail指向的位置上elems[tail] = elem;tail++;// 如果队尾大于数组大小，让tail重新指向0下标，形成闭环if (tail >= elems.length) { // tail = tail % elems.lengthtail = 0;}size++;// 入队列成功后进行唤醒locker.notify();}}public String take() throws InterruptedException {String elem = null;synchronized (locker) {while (size == 0) {//队列空了// 实现阻塞locker.wait();}// 取出 head 位置的元素并返回elem = elems[head];head++;if (head >= elems.length) {head = 0;}size--;// 队列不满，即出队列成功之后，加上唤醒locker.notify();}return elem;}}
public class ThreadDemo29 {public static void main(String[] args) throws InterruptedException {MyBlockingQueue queue = new MyBlockingQueue(1000);// 生产者消费者模型（核心是阻塞队列,使用synchronized和wait/notify达到线程安全&阻塞） 不仅仅是一个线程// 也可能是一个独立的服务器程序，甚至是一组服务器程序// 生产者Thread t1 = new Thread(() -> {int n = 1;while (true) {try {queue.put(n+" ");System.out.println("生产元素：" + n);n++;//Thread.sleep(500); //生产一个消费一个} catch (InterruptedException e) {throw new RuntimeException(e);}}});// 消费者Thread t2 = new Thread(() -> {while (true) {try {String n = queue.take();System.out.println("消费元素：" + n);Thread.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});t1.start();t2.start();}
}

3.定时器

在Java标准库中提供了定时器的使用，Timer类，核心方法是schedule()，它有两个参数，第一个参数是即将执行的任务代码，第二个是指定多长时间之后执行（单位ms）。

3.1 定时器的使用

import java.util.Timer;
import java.util.TimerTask;
// 运行完之后，进程没有结束，因为timer 里内置了线程（前台线程） timer不知道是否还要添加任务进来，
// 可以使用timer.cancel()来主动结束public class ThreadDemo30 {public static void main(String[] args) throws InterruptedException {// 定义一个timer添加多任务，每个任务同时会带有一个时间Timer timer = new Timer();timer.schedule(new TimerTask() {@Overridepublic void run() {//时间到了之后，要执行的代码System.out.println("hello timer 3000");}},3000);timer.schedule(new TimerTask() {@Overridepublic void run() {System.out.println("hello timer 2000");}},2000);timer.schedule(new TimerTask() {@Overridepublic void run() {System.out.println("hello timer 1000");}},1000);System.out.println("hello main");Thread.sleep(3000);timer.cancel();}
}

3.2 实现定时器

• 一个带优先级的阻塞队列（按从小到大的顺序）
• 队列中的每个元素是一个Task对象
• Task中带有一个时间属性，队首元素就是即将执行的
• 同时有一个worker线程一直扫描队首元素，看队首元素是否需要执行

import java.util.Comparator;
import java.util.PriorityQueue;
import java.util.Timer;// 通过这个类，来描述一个任务
class MyTimerTask implements Comparable<MyTimerTask> {//在什么时间点来执行这个任务// 此处约定这个time是一个ms 级别的时间戳private long time;public long getTime() {return time;}// 实际任务要执行的代码private Runnable runnable;// delay 期望是一个相对时间public MyTimerTask(Runnable runnable, long delay) {this.runnable = runnable;// 计算真正要执行任务的绝对时间（）this.time = System.currentTimeMillis() + delay;}public void run() {runnable.run();}@Overridepublic int compareTo(MyTimerTask o) {return (int) (this.time - o.time);}
}// 通过这个类，来表示一个定时器
class MyTimer {// 负责扫描任务队列，执行任务的线程private Thread t = null;// 任务队列private PriorityQueue<MyTimerTask> queue = new PriorityQueue<>();private Object locker = new Object();// 把任务放进队列public void schedule(Runnable runnable,long delay) {synchronized (locker) {MyTimerTask task = new MyTimerTask(runnable,delay);queue.offer(task);// 添加新的元素之后，就可以唤醒扫描线程的wait了locker.notify();}}public void cancel() {//}// 构造方法，创建扫描线程，让扫描线程来完成判定和执行public MyTimer() {t = new Thread(() -> {// 扫描线程就需要循环反复的扫描队首元素，然后判定队首元素是不是时间到了// 如果没到时间，啥也没有// 如果时间到了，就执行这个任务从队列中删除while (true) {try {// 1. 解决线程安全问题synchronized (locker) { //这里的代码执行速度很快，解锁之后立即又重新尝试加锁，导致// 其他线程通过schedule想加锁，但是加不上 （即线程饿死） -》 引入wait/notifyif (queue.isEmpty()) {// 暂时先不处理locker.wait();}MyTimerTask task = queue.peek();// 获取当前时间long curTime = System.currentTimeMillis();if (curTime >= task.getTime()) {//当前时间已经达到了任务时间，就可以执行任务了queue.poll();task.run();}else {// 当前时间还没到任务时间，暂时不执行// 不能使用sleep。会错过新的任务，也无法释放锁//Thread.sleep(task.getTime() - curTime);locker.wait(task.getTime() - curTime);}}  // 释放锁的} catch (InterruptedException e) {e.printStackTrace();}}});// 启动线程t.start();}
}
public class ThreadDemo31 {public static void main(String[] args) {MyTimer timer = new MyTimer();timer.schedule(new Runnable() {@Overridepublic void run() {System.out.println("hello 3000");}},3000);timer.schedule(new Runnable() {@Overridepublic void run() {System.out.println("hello 2000");}},2000);timer.schedule(new Runnable() {@Overridepublic void run() {System.out.println("hello 1000");}},1000);System.out.println("hello main");}
}

4 线程池

线程池：提前把需要用的线程，在线程池里准备好，需要用的时候就从池子里取，用完之后还给池子。
由于频繁的创建销毁进程，成本太高，引入了轻量级 进程，即线程，如果创建销毁线程的频率也进一步提高，此时线程的创建销毁开销也越来越大。
所以有两种优化此处的线程的创建和销毁：

1. 引入轻量级 线程，即纤程/协程：本质是程序员在用户太代码进行调度，而不是靠内核的调度器调度，节省了调度上的开销。
2. 线程池：把要使用的线程提前创建好，用完了之后也不要直接释放，而是放进线程池里以备下次使用，从而节省了创建销毁线程的开销。

引入问题：为什么从线程池里取线程就比系统申请更高效？

• 从线程池里取线程是纯用户态代码（可控的）
• 通过系统申请创建线程，就是需要内核来完成（不太可控）

4.1 线程池的使用

在Java标准库中，把ThreadPoolExecutor类表线程池，给封装 成 Executors 工厂类，工厂类：创建出不同的线程池对象（在内部把ThreadPoolExecutor创建好了并且设置不同的参数）。
ThreadPoolExecutor 线程池的参数：

• int corePoolSize :核心线程数，int maximunmPoolSize：最大线程数
• long keepAliveTime：保持存活时间，TimeUnit unit:时间单位（s,min,ms,hour）
• ThreadFactory threadFactor：线程工厂，通过这个工厂类来创建线程对象（Thread）
• RejectExecutionHandler handler:拒绝策略，在线程池中，有一个阻塞队列，能够容纳的元素有上限，当任务队列已经满了，如果继续往队列添加任务，线程池会进行下面4种操作：

Executor创建线程池的方式：

• newFixedThreadPool: 创建固定线程数的线程池
• newCachedThreadPool: 创建线程数目动态增长的线程池.
• newSingleThreadExecutor: 创建只包含单个线程的线程池.
• newScheduledThreadPool: 设定 延迟时间后执行命令，或者定期执行命令. 是进阶版的 Timer.

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class ThreadDemo32 {public static void main(String[] args) {// 创建线程池的时候，设定线程池的线程数量ExecutorService service = Executors.newFixedThreadPool(4);service.submit(new Runnable() {@Overridepublic void run() {System.out.println("hello");}});}
}

** 创建线程池的时候，很多时候需要设定线程池的线程数量？**

• 不同的程序，能够设定的线程的数目是不同的，要具体问题具体分析， 一个线程是CPU密集型的任务（在线程run里面进行计算），还是IO密集型任务（在线程run里使用scanner读取用户的输入）
• 如果一个进程中所有的线程都是CPU密集型的，每个线程所有工作都是CPU上执行，此时线程数目就不应该超过N（CPU的逻辑核心数）。
• 如果一个进程中，所有线程都是IO密集型，每个线程的大部分工作都是等待IO，此时线程数目与那元超过N
• 由于程序的复杂性，所以需要通过实验/测试，即设定不同的线程数目，分别进行性能测试，衡量每种线程数目下，总的时间开销，和系统资源占用的开销，找到这之间的合适值。

4.2 实现一个简单的线程池 – 重点

1. 提供构造方法，指定创建多少个线程
2. 在构造方法中，把这些线程都创建好
3. 有一个阻塞队列，能够持有要执行的任务
4. 提供submit方法，可以添加新的任务

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;class MyThreadPoolExecutor {private List<Thread> threadList = new ArrayList<>();// 保存任务的队列private BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1000);// 通过n来指定创建多少个线程public MyThreadPoolExecutor(int n) {for (int i = 0; i < n; i++) {Thread t = new Thread(() -> {// 线程把任务队列中的任务不停的取出来，并且进行执行while (true) {try {// 此处的take 带有阻塞功能// 如果队列为空，此时 take就会阻塞Runnable runnable = queue.take();// 取出一个执行一个runnable.run();} catch (InterruptedException e) {e.printStackTrace();}}});t.start();threadList.add(t);}}public void submit(Runnable runnable) throws InterruptedException {queue.put(runnable);}
}public class ThreadDemo33 {public static void main(String[] args) throws InterruptedException {MyThreadPoolExecutor executor = new MyThreadPoolExecutor(4);for (int i = 0; i < 1000; i++) {// n是一个实事final变量 ，每次循环都是一个新的n，就可以被捕获int n = i;executor.submit(new Runnable() {@Overridepublic void run() { // 回调函数访问当前外部作用域的变量就是变量捕获// i 一值在变，把i改成成员变量或者 int n = iSystem.out.println("执行任务" + n + ",当前线程为："+ Thread.currentThread().getName());}});}}
}