JavaEE：多线程（3）：案例代码

目录

案例一：单例模式

饿汉模式

懒汉模式

思考：懒汉模式是否线程安全？

案例二：阻塞队列

可以实现生产者消费者模型

削峰填谷

接下来我们自己实现一个阻塞队列

1.先实现一个循环队列

2. 引入锁，实现线程安全

3.实现阻塞

实现生产者消费者模型

案例三：定时器

问题

线程安全

线程饿死

理解代码过程

案例四：线程池

标准库中的线程池：ThreadPoolExecutor

Executors工厂类

手敲线程池

---

多线程基础知识要点

案例一：单例模式

是一种设计模式

软件设计需要框架，这是硬性的规定；设计模式是软性的规定。遵循好设计模式，代码的下限就被兜住了

单例 = 单个实例（对象）

某个类在一个进程中只应该创建出一个实例（原则上不应该有多个）

使用单例模式可以对代码进行一个更严格的校验和检查

实现单例模式~

饿汉模式

第1步：

class Singleton{
    private static Singleton instance = new Singleton();
}

这里的static指的是类属性，而instance就是Singleton类对象持有的属性

每个类的类对象只存在一个，类对象中的static属性自然只有一个了

因此instance指向的这个对象，就是唯一的对象

第2步：

其他代码要想使用这个类的实例就需要通过这个方法来进行获取。不应该在其他代码钟重新new这个对象，而是使用这个方法获取到现成的对象（已经创建好的对象）

第3步：奇淫巧计

这里直接把Singleton给private了，其他代码根本没办法new

此时，无论你创建多少个对象，这些对象其实都是一样的

饿汉模式下，实例是在类加载的时候就创建了，创建时机非常早，相当于程序一启动，实例就创建了。“饿汉”形容“创建实例非常迫切，非常早”

欸！但是用非常规手段：反射就可以打破上述约定。我们可以用枚举方法来创建单例模式

---

懒汉模式

创建实例的时机更晚，只到第一次使用的时候才会创建实例

“懒”的思想

比如有一个非常大的文件(10GB)，有一个编辑器，使用编辑器打开这个文件，如果是按照饿汉的方式，编辑器就要把这10GB先加载到内存里，然后再统一地展示。加载太多数据，用户还得一点点看，没办法一下看那么多

如果按照懒汉的方式，编辑器就会制度取一小部分数据，把这部分数据先展示出来，随着用户翻页之类的操作再继续读后面的数据。这样效率可以提高

class SingletonLazy{
    private static SingletonLazy instance = null;//先初始化为null，不是立即初始化
    public static SingletonLazy getInstance(){
        if (instance == null){
            instance = new SingletonLazy();//首次调用getInstance才会创建出一个实例
        }
        return instance;
    }
    private SingletonLazy(){}
}

思考：懒汉模式是否线程安全？

这样t1 new了一个对象，t2也new了一个对象，就会出现bug

所以，懒汉模式不是线程安全的

那怎么改成线程安全的呢？

1.加锁，synchronized

2.把if和new两个操作打包成一个原子

仍然是t1和t2两个线程，t1先执行加锁代码，t2就被阻塞了，要等待t1释放锁才能继续执行

而t1把instance修改之后，t2的if条件就不成立了，直接就返回了

---

emm，这段代码还不够完美...

在多线程里面，当第一个线程加了锁，后面的线程再调用getInstance就是纯粹的读操作了，也就不会有线程问题了。那么没有线程的代码每次执行都要加锁和解锁，每次都会产生阻塞，效率巨低！

所以在synchronized外边还得再套一层if，判定代码是否要加锁。仍然将instance是否为空作为判断条件

第一个if判定是否加锁

第二个if判定是否要创建对象 

---

🆗上面的代码还有一点问题

涉及到指令重排序引起的线程安全问题

指令重排序是指调整原有代码的执行顺序，保证逻辑不变的前提下提高程序的效率

为什么调整代码执行顺序可以提高程序效率？

比如我们去超市买东西，我们需要买黄瓜，胡萝卜，西红柿，土豆。我们就有很多种去不同摊位的路径选择，每种选择的最终总购买时间不一样。这就相当于程序的效率。

这行代码可以分成三个大步骤

1.申请一段内存空间

2.在这个内存上调用构造方法，创建出这个实例

3.把这个内存地址赋给Instance引用变量

假设有t1和t2两个线程

t1线程按照1 3 2的执行顺序，就会出现问题

解决上述问题核心思路：volatile

volatile有两个功能：

1）保证内存可见性，每次访问变量必须要重新读取内存，而不会优化到寄存器/缓存中

2）禁止指令重排序，针对这个volatile修饰的变量的读写操作的相关指令，是不能被重排序的

这样修改之后，针对instance变量的读写操作就不会出现重排序

---

案例二：阻塞队列

特点：1.线程安全；2.阻塞

如果一个已经满了的队列进行入队列，此时入队列操作就会阻塞，一直阻塞到队列不满之后

如果一个已经空的队列进行出队列，出队列操作就会阻塞，一直阻塞到队列有元素为止

可以实现生产者消费者模型

这个模型可以更好地解耦合(把代码的耦合程度从高降低)

实际开发中，往往会用到分布式系统，服务器整个功能不是由一个服务器完成的，而是每个服务器负责一部分功能。通过服务器之间的网络通信，最终完成整个功能

在这个案例中，A和B,C之间的耦合性比较强，一旦B或者C挂了一个，A也就跟着挂了

如果引入生产者消费者模型

这个阻塞队列不是简单的数据结构，而是基于这个数据结构实现的服务器程序，又被部署到单独的主机上

---

削峰填谷

为啥当请求多了的时候，服务器就容易挂？

因为服务器处理每个请求都是要消耗硬件资源（包括但不限于CPU，内存，硬盘，网络带宽），上述任何一种硬件资源达到瓶颈，服务器都会挂

因为B和C抗压能力比较弱，所以我们可以用一个阻塞队列来承担峰值请求

阻塞队列：数据结构

消息队列：基于阻塞队列实现服务器程序

Java标准库里线程的阻塞队列

BlickingQueue: ArrayBlockingQueue, LinkedBlockingQueue, PriorityBlockingQueue

        BlockingDeque queue = new ArrayBlockingQueue<>(100);
        queue.put("aaa");

put和offer都是入队列，但是put带有阻塞功能，而offer没带阻塞功能，队列满了会返回布尔结果

        String elem = queue.take();
        System.out.println("elem = "+elem);

take用来出队列，带有阻塞功能 

---

接下来我们自己实现一个阻塞队列

1.先实现一个循环队列

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];
    }
 
    public void put(String elem){
        //新的元素放到tail指向的位置上
        if (size >= elems.length){
            //队列满了，需要下面这个代码阻塞
            return;
        }
        //新的元素要放到tail指向的元素上
        elems[tail] = elem;
        tail++;
        if(tail >= elems.length){
            tail = 0;
        }
        size++;
    }
    public String take(){
        if(size == 0){
            //队列空了，需要下面这个代码阻塞
            return null;
        }
        String elem = elems[head];
        head ++;
        if(head >= elems.length){
            head = 0;
        }
        size--;
        return null;
    }
}

2. 引入锁，实现线程安全

    private static Object locker = new Object();
    public MyBlockingQueue(int capacity){
        elems = new String[capacity];
    }
 
    public void put(String elem){
        synchronized (locker){
            //新的元素放到tail指向的位置上
            if (size >= elems.length){
                //队列满了，需要下面这个代码阻塞
                return;
            }
            //新的元素要放到tail指向的元素上
            elems[tail] = elem;
            tail++;
            if(tail >= elems.length){
                tail = 0;
            }
            size++;
        }
    }
    public String take(){
        String elem = null;
        synchronized (locker){
            if(size == 0){
                //队列空了，需要下面这个代码阻塞
                return null;
            }
            elem = elems[head];
            head ++;
            if(head >= elems.length){
                head = 0;
            }
            size--;
            return elem;
        }
    }
}

3.实现阻塞

对于满了的情况，用wait方法阻塞，在出队列成功之后再进行唤醒

队列空的情况，在入队列成功后的线程中唤醒

    public void put(String elem) throws InterruptedException {
        synchronized (locker){
            //新的元素放到tail指向的位置上
            while (size >= elems.length){
                //队列满了，需要下面这个代码阻塞
                locker.wait();
            }
            //新的元素要放到tail指向的元素上
            elems[tail] = elem;
            tail++;
            if(tail >= elems.length){
                tail = 0;
            }
            size++;
            //入队列成功后唤醒
            locker.notify();
        }
    }
    public String take() throws InterruptedException {
        String elem = null;
        synchronized (locker){
            while (size == 0){
                //队列空了，需要下面这个代码阻塞
                locker.wait();
            }
            elem = elems[head];
            head ++;
            if(head >= elems.length){
                head = 0;
            }
            size--;
            //出队列成功后唤醒
            locker.notify();
        }
        return elem;
    }

这里的if为什么改成while了呢？

因为if只能判定一次条件，有时候一旦程序进入阻塞之后再被唤醒，中间隔的时间会很长，这个间隔过程变数很多，可能这个入队列的条件无法再满足了。

欸那改成while之后，就是wait唤醒之后再判定一次条件，wait之前判定一次，唤醒之后再判定一次（就是多做一次确定）。再次确认发现队列还是满的，那就继续等待。

---

实现生产者消费者模型

    public static void main(String[] args) {
        MyBlockingQueue queue = new MyBlockingQueue(1000);
        //生产者
        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);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        t1.start();
        t2.start();
    }

实际开发中，生产者和消费者往往不仅仅是一个线程，也可能是一个独立的服务器程序

---

案例三：定时器

可以设定一个时间，时间到了的时候，定时器自动执行某个逻辑（比如，写博客定时发布）

用法：定义一个timer添加多个任务，每个任务同时会带有一个时间

Timer里面内置了前台线程，因为timer不知道你的代码是否还会添加新的任务进来，仍然严正以待

需要使用cancel来主动结束

---

现在我们来手搓一个定时器

需要有什么？1.一个可以帮我们掐时间的线程；2.一个能帮我们存储任务的优先级队列

因为每个任务都带有delay时间的，用优先级队列可以先执行时间小的，后执行时间大的

扫描线程就不必遍历了，只需要关注队首元素是否到时间（队首没到时间，其他元素也没到时间）

计时任务

任务优先级逻辑（时间小的优先级越高）

计时器

---

问题

线程安全

由于我们在主线程中对队列的元素进行添加，而扫描线程对已完成的元素进行删除，两个线程操作同一个优先级队列变量，会有线程安全问题

此时需要加锁来解决线程安全问题

schedule方法（主线程要调用）的加锁

---

 判断：以下哪种加锁方法是正确的

   //第一种
     public MyTimer(){
        t = new Thread(()->{
            //扫描线程就需要循环的反复扫描队首元素，然后判定队首任务时间是否到达
            //时间到了就执行任务并删除这个任务
            //时间没到就啥都不干
            synchronized (locker){
                while (true){
                    if (queue.isEmpty()){
                        continue;
                    }
                    MyTimerTask task = queue.peek();
                    //获取当前时间
                    long curTime = System.currentTimeMillis();
                    if(curTime >= task.getTime()){
                        //当前时间已经到了任务时间，就可以执行任务了
                        queue.poll();
                        task.run();
                    }else{
                        //时间还没到，暂时先不执行
                        continue;
                    }
                } 
            }
        });
 
   //第二种
    public MyTimer(){
        t = new Thread(()->{
            //扫描线程就需要循环的反复扫描队首元素，然后判定队首任务时间是否到达
            //时间到了就执行任务并删除这个任务
            //时间没到就啥都不干
            
            while (true){
                synchronized (locker){
                    if (queue.isEmpty()){
                        continue;
                    }
                    MyTimerTask task = queue.peek();
                    //获取当前时间
                    long curTime = System.currentTimeMillis();
                    if(curTime >= task.getTime()){
                        //当前时间已经到了任务时间，就可以执行任务了
                        queue.poll();
                        task.run();
                    }else{
                        //时间还没到，暂时先不执行
                        continue;
                    }
                }
            }
        });
    }

第一种方法，把锁放到while外面，如果while没有结束的话，锁永远都释放不了，主线程调用schedule方法就永远上不了锁。所以我们要采用第二种方法，把锁加到while里面，才有释放锁的机会

---

线程饿死

上面的第二种方法虽然解决了线程安全问题，但是这部分代码执行速度很快，解锁之后就立即重新加锁，导致其他线程想通过schedule加锁都加不上，所以我们需要使用wait来解决

一旦由新的任务加入，wait就会被唤醒，因为不知道加入的任务是不是最早的任务，所以我们用task.getTime() - curTime来获取任务时间

没有新的任务，时间到了。按照原定计划，执行之前的这个最早的任务即可

执行结果

---

理解代码过程

理解peek:

⚠

优先级队列：无论添加多少元素，这里的peek都是得到时间最小的值。

理解run方法

👇

👇

👇（Runnable作为描述任务的主体）

👇

main方法里面写出任务具体执行代码

---

案例四：线程池

池是什么？

池就相当于一个共享资源，是对资源的整合和调配，节省存储空间，当需要的时候可以直接在池中取，用完之后再还回去。比如，如果你喝水，你可以拿杯子去水龙头接。如果很多人喝水，那就只能排队去接。

Java常用的池有常量池，数据库连接池，线程池，进程池，内存池

最开始进程能够解决并发编程的问题，但是因为频繁创建销毁进程的成本太高了，引入了线程这种轻量级进程。但是如果创建销毁线程的频率进一步提高，这里的开销也不能忽视

那怎么优化线程创建销毁效率呢？

1.引入轻量级线程--纤程/协程

协程本质是程序员在用户态代码中进行调度，不是靠内核的调度器来调度的

协程运行在线程之上，当一个协程执行完成后，可以选择主动让出，让另一个协程运行在当前线程之上。可以节省很多调度上的开销。

⚠线程里有协程这句话是不严谨的，因为协程本身不是系统级别的概念，是用户代码中基于线程封装出来的，有不同的实现方法，可能n个协程对应1个线程，也可能n个协程对应m个线程

2.引入线程池。把要使用的线程提前创建好，用完了也不要直接释放而是备下次使用，就节省创建/销毁线程的开销

从线程池里取线程（纯用户态代码）比从系统申请更高效！

比如一个事情一个人自己就能完成，就更可控，更高效，这种相当于纯用户态代码

但是如果这个事情这个人要拜托其他人来完成，不知道委托人要花多少时间，就不可控，更低效。相当于去系统申请线程

---

标准库中的线程池：ThreadPoolExecutor

构造方法（面试题常考）

标准库提供的线程池，持有的线程个数并不是一成不变的，会根据当前的任务量自适应线程个数 

 核心线程数（规定一个线程池里最少有多少个线程）

最大线程数（规定一个线程最大有多少个线程）

某个线程超过保持存活时间阈值就会被销毁掉

和定时器类似，线程池中可以持有多个任务

 -- 线程工厂

通过这个工厂类创建线程对象（Thread对象），在这个类里面提供了方法，让方法封装new Thread的操作，同时给Thread设置一些属性

设计模式：工厂模式。通过专门的工厂类/对象来创建指定的对象

例子：

//平面上的一个点
 
class Point{
 
        public Point(double x, double y){...}//通过笛卡尔坐标构造这个点
 
        //还可以用三角函数转换笛卡尔坐标
 
        //x = r * cos(a); y = r * sin(a)
 
        public Point(double r, double a){...}//通过极坐标系来构造点（半径，角度）
 
}

上面代码能编译通过吗？不能。因为不能构成重载（因为形参类型和个数相同了）

为了让上面代码通过，就可以引入工厂模式

class Point{
    //工厂方法
    public static Point makePointByXY(double x, double y){
        Point p = new Point();
        p.setX(x);
        p.setY(y);
        return p;
    }
    public static Point makePointByRA(double r, double a){
        Point p = new Point();
        p.setR(r);
        p.setA(a);
        return p;
    }
    Point p = Point.makePointByXY(x, y);
    Point p = Point.makePointByRA(r, a);
}

通过静态方法封装new操作，在方法内部设定不同的属性完成对象初始化，这个构造对象的过程就是工厂模式

拒绝策略

在线程池中，有一个阻塞队列，能够容纳的元素有上限，当任务队列已经满了，如果继续往队列中添加任务，线程池中就会拒绝添加

四种拒绝策略

第一种：继续添加任务，直接抛出异常

第二种：新的任务由添加任务的线程负责执行（线程池不会执行）--谁揽的活谁干

第三种：丢弃最老的任务

第四种：丢弃最新的任务

---

Executors工厂类

通过这个类创建不同的线程池对象

例子

    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(4);
        service.submit(new Runnable() {
            @Override
            public void run() {
 
            }
        });
    }

啥时候使用Executors，啥时候使用ThreadPoolExecutor

Executors方便只是简单用一下，ThreadPoolExecutor希望高度定制化

---

线程池里最好有多少个线程？（具体情况具体分析，回答具体数字就是错误的）

线程里的任务分成两种

CPU密集型任务：这个线程大部分时间都在CPU上运行/计算。比如在线程run里面计算1+2+3+...+10w。

IO密集型任务：这个线程大部分时间都在等待IO，不需要去CPU上运行。比如线程run里加scanner，读取用户输入。

如果一个进程中，所有的线程都是CPU密集型，每个线程所有的工作都在CPU上执行。此时，线程数目就不应该超过N（CPU逻辑核心数）。——每个线程都要占一个核，超过N就失控了

如果一个进程中，所有的线程都是IO密集型，每个线程大部分工作都在等待IO，CPU消耗非常少。此时线程数目就可以很多，远远超过N。——一个线程工作，其他线程休息，不霸占CPU核

---

手敲线程池

1.提供构造方法，指定创建多少个线程

2.在构造方法中，把这些线程都创建好

3.有一个阻塞队列，能够持有要执行的任务

4.提供submit方法，能够添加新的任务

写的过程中遇到问题

run变量捕获到i之后，正常情况i是不能变的，但是i因为循环造成改变，引发编译器异常

此处的n就是一个实时final变量，每次循环就创建一个不可变的n，这个n是可以被捕获的

package Thread;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
 
class MyThreadPoolExecutor{
    private List threadList = new ArrayList<>();
    //创建一个用来保存任务的队列
    private BlockingQueue 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 ThreadDemo11 {
    public static void main(String[] args) throws InterruptedException{
        MyThreadPoolExecutor executor = new MyThreadPoolExecutor(4);
        for (int i = 0; i < 1000; i++) {
            int n = i;
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println("执行任务 "+ n + ", 当前线程为：" + Thread.currentThread().getName());
                }
            });
        }
    }
}

更体现多线程执行顺序不确定