六、并发编程

怼烎@ 2022-11-06 04:58 218阅读 0赞

# 1. 线程 #

线程是操作系统能够进行运算调度的最小单位，它被包含在进程之中，是进程中的实际运作单位。

线程出现之前，任务的执行是以进程为单位的。

## **1.1 线程的作用** ##

*  在多核CPU中，利用多线程可以实现真正意义上的并行执行。例如当前系统是八核CPU，那么意味着可以同时运行8个线程。
 *  在一个应用进程中，会存在多个同时执行的任务，如果其中一个任务被阻塞，将会引起不依赖该任务的任务也被阻塞。通过对不同任务创建不同线程去处理，可以提升程序处理的实时性。
 *  线程可以认为是轻量级的进程，线程的创建、销毁比进程更快。

## 1.2 多线程 ##

**为什么要用多线程？**

*  异步执行
 *  利用多核CPU资源实现真正意义上的并行执行
 *  单线程是串行执行，而多线程是多任务并发执行，减少耗时，充分利用CPU资源

**多线程的应用场景**

*  使用多线程实现文件下载
 *  后台任务：如定时向大量用户发送邮件
 *  异步处理：如记录日志
 *  多步骤任务处理：可根据步骤特征选择不同个数和特征的线程来协作处理，多任务分割，由一个主线程分割给多个线程完成

总结：多线程的本质是合理利用多核CPU资源来实现线程的并行处理，来实现同一个进程内的多个任务并行执行，同时基于线程本身的异步执行特性，提升任务处理效率。

## **1.3 Java中创建线程的方式** ##

*  继承Thread类
 *  实现Runnable接口
 *  实现Callable接口：如果需要线程执行完得到一个返回值，选择此方式

### 1.3.1 继承Thread类 ###

public class ThreadDemo extends Thread {
        @Override
        public void run() {
            System.out.println("This is my customized thread.");
        }
    
        public static void main(String[] args) {
            ThreadDemo threadDemo = new ThreadDemo();
            threadDemo.start();
        }
    }

### 1.3.2 实现Runnable接口 ###

public class RunnableDemo implements Runnable {
        @Override
        public void run() {
            System.out.println("This is my customized thread.");
        }
    
        public static void main(String[] args) {
            RunnableDemo runnableDemo = new RunnableDemo();
            Thread thread = new Thread(runnableDemo);
            thread.start();
        }
    }

### 1.3.3 实现Callable接口 ###

public class CallableDemo implements Callable<String> {
        @Override
        public String call() throws Exception {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            return "Hello, this is response value.";
        }
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            CallableDemo callableDemo = new CallableDemo();
            ExecutorService executorService = Executors.newFixedThreadPool(1);
            Future<String> future = executorService.submit(callableDemo);
            // Future.get()是一个阻塞方法，会阻塞主线程
            System.out.println(Thread.currentThread().getName() + "########" + future.get());
        }
    }

**Future/Callable实现原理**

任务执行不会阻塞，但是get()获取返回结果时，如果没有得到返回值会阻塞等待。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70][]

## 1.4 线程的生命周期 ##

Java线程从创建到销毁，一共可能有6种状态：

> **NEW**：初始状态，线程被创建，但是还没有调用start方法  
> **RUNNABLE**：运行状态，有可能在jvm中执行，也可能等待OS进程去处理  
> **BLOCKED**：阻塞状态，表示线程进入等待状态，线程因为某种原因放弃了CPU使用权  
> **WAITING**：等待状态  
> **TIMED\_WAITING**：超时等待状态，超时以后自动返回  
> **TERMINATED**：终止状态，当前线程执行完毕

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1][]

## 1.5 线程的基本操作及原理 ##

### 1.5.1 Thread.join()的使用及原理 ###

案例

public class ThreadJoinDemo {
        private static int a = 0, b = 0;
        public static void main(String[] args) throws InterruptedException {
            Runnable runnable = new Runnable() {
                @Override
                public void run() {
                    a = 1;
                    b = 2;
                }
            };
            Runnable runnable2 = new Runnable() {
                @Override
                public void run() {
                    a = a + b;
                }
            };
            Thread thread1 = new Thread(runnable); //创建线程1
            thread1.start();
            Thread thread2 = new Thread(runnable2); //创建线程2
            thread2.start();
            Thread.sleep(1000); //先sleep 1秒，保证两个线程都已经start
            System.out.println("print result: a = " + a);
        }
    }

我们可以发现一件很有趣的事，打印结果a可能是1可能是3

![20210309152340110.png][]![20210309152405927.png][]

那么为什么会这样呢？因为两个线程同时运行，我们无法保证两个线程的执行顺序

所以如果想要保证第一个线程先执行，第二个线程基于第一个线程的执行结果再执行，那么就可以使用**join()**方法

Thread thread1 = new Thread(runnable);
    thread1.start();
    thread1.join(); //保证thread1比thread2先运行，阻塞主线程，直到thread1运行结束
    Thread thread2 = new Thread(runnable2);
    thread2.start();
    Thread.sleep(1000);
    System.out.println("print result: a = " + a);

主线程、线程1、线程2执行流程图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 2][]

Thread.join()方法原理和源代码分析

public final void join() throws InterruptedException {
        join(0);
    }
    public final synchronized void join(long millis) throws InterruptedException {
        long base = System.currentTimeMillis();
        long now = 0;
        if (millis == 0) { //join方法进入if
            while (isAlive()) { //主线程处于存活状态
                wait(0); //执行wait方法阻塞主线程
            }
        } else {
            while (isAlive()) {
                long delay = millis - now;
                if (delay <= 0) {
                    break;
                }
                wait(delay);
                now = System.currentTimeMillis() - base;
            }
        }
    }

### 1.5.2 Thread.sleep()的使用 ###

使线程暂停执行一段时间，直到等待的时间结束才恢复执行或在这段时间内被中断

Thread.sleep(3000); //睡眠3秒

Thread.sleep()的工作流程：

*  挂起线程并修改其运行状态为阻塞状态
 *  用sleep(3000)提供的参数来设置一个定时器
 *  当时间结束，定时器会被触发，内核收到中断后修改线程的运行状态为RUNNABLE，进入队列等待OS调度

--------------------

**线程调度算法**

在操作系统中，CPU竞争有多种策略。Unix系统使用的时间片轮转算法，而Windows则使用的抢占式算法。

时间片轮转（RR）调度算法：将一个较小时间单元定义为时间量或时间片。时间片的大小通常为 10~100ms。就绪队列作为循环队列。CPU 调度程序循环整个就绪队列，为每个进程分配不超过一个时间片的 CPU。如果进程阻塞或提前执行完毕，进程会自动释放 CPU。调度程序接着处理就绪队列的下一个进程。如果一个时间片结束，当前进程还没有执行完毕，那么会进程切换，执行就绪队列中的下一个进程，将当前未执行完毕的进程加到就绪队列的尾部。

抢占式调度算法：如果进程抢到了CPU资源，那么就会长期霸占，直到任务执行结束。

--------------------

**问题思考：**

Thread.sleep(3000)睡眠3秒后会立马执行吗？  
实际上，睡眠3秒后才参与到CPU竞争中，不一定立马就会执行，直到竞争中获取到CPU资源才会执行。

Thread.sleep(0)的意义？  
放弃本次CPU资源的使用权，要求操作系统重新触发CPU竞争，重新分配资源。

### 1.5.3 Thread.wait()和notify()的使用 ###

wait()方法阻塞线程，notify()方法唤醒线程。

**线程的通信：**线程A修改一个值，线程B可以知道修改后的结果

> **线程通信机制**
> 
>  *  线程的通信就是根据wait和notify来实现
>  *  对共享变量进行抢占，抢占同一把锁Lock
>  *  wait和notify必须在同一个synchronized中

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 3][]

wait()方法和notify()方法可以实现一个简单的生产者-消费者模型。

public class Producer implements Runnable {
        private int size;
        private Queue<String> products;
    
        public Producer(int size, Queue<String> products) {
            this.size = size;
            this.products = products;
        }
    
        @Override
        public void run() {
            int i = 0;
            while (true) {
                i++;
                synchronized (products) {
                    while (products.size() == size) {
                        System.out.println("仓库已满，无法再继续生产products...");
                        try {
                            products.wait(); //阻塞
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("生产者生产了一个product"+i);
                    products.add("product" + i);
                    products.notify(); //唤醒消费者消费
                }
            }
        }
    }

public class Consumer implements Runnable {
        private int size;
        private Queue<String> products;
    
        public Consumer(int size, Queue<String> products) {
            this.size = size;
            this.products = products;
        }
    
        @Override
        public void run() {
            while (true) {
                synchronized (products) {
                    while (products.size() == 0) {
                        System.out.println("仓库是空的...");
                        try {
                            products.wait(); //阻塞
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    String product = products.remove();
                    System.out.println("消费者消费了一个" + product);
                    products.notify(); //唤醒生产者生产
                }
            }
        }
    }

public class WaitNotifyDemo {
        public static void main(String[] args) {
            int size = 10;
            Queue<String> queue = new LinkedList<String>();
            Producer producer = new Producer(size, queue);
            Consumer consumer = new Consumer(size, queue);
            Thread thread1 = new Thread(producer);
            Thread thread2 = new Thread(consumer);
            thread1.start(); //生产者线程启动
            thread2.start(); //消费者线程启动
        }
    }

输出结果如下：

生产者生产了一个product1
    生产者生产了一个product2
    生产者生产了一个product3
    生产者生产了一个product4
    消费者消费了一个product1
    消费者消费了一个product2
    消费者消费了一个product3
    生产者生产了一个product5
    生产者生产了一个product6
    生产者生产了一个product7
    生产者生产了一个product8
    生产者生产了一个product9
    生产者生产了一个product10
    生产者生产了一个product11
    生产者生产了一个product12
    生产者生产了一个product13
    仓库已满，无法再继续生产products...
    消费者消费了一个product4
    生产者生产了一个product14
    消费者消费了一个product5
    消费者消费了一个product6
    消费者消费了一个product7
    消费者消费了一个product8
    消费者消费了一个product9
    生产者生产了一个product15
    消费者消费了一个product10
    消费者消费了一个product11
    消费者消费了一个product12
    消费者消费了一个product13
    消费者消费了一个product14
    消费者消费了一个product15
    生产者生产了一个product16
    消费者消费了一个product16
    仓库是空的...
    生产者生产了一个product17
    生产者生产了一个product18
    生产者生产了一个product19
    生产者生产了一个product20
    生产者生产了一个product21
    生产者生产了一个product22
    ...

为什么在使用wait()和notify()方法时都加上了**synchronized**关键字？

a. 因为wait()和notify()两个方法之间是互斥的，所以需要synchronized来锁住对象，wait和notify执行时首先需要竞争获取同步锁，让对方无法同时执行。  
b. 因为wait()和notify()用于多个线程之间互相通信，那么通信需要一个通信的载体，而synchronized就是这个载体wait和notify是基于synchronized来实现通信的，两者必须要在同一个频道也就是同一个锁的范围之内。

### 1.5.4 如何正确的中断一个线程执行？ ###

正确的友好的中断一个线程，就要保证线程中的run方法正常执行完毕，不会影响执行结果。

**thread.stop()**方法可以中断线程，但是已经不推荐使用，因为这种方式是强制终止，是不友好的，是在jvm层面终止了线程，会对线程执行结果有影响，即使线程的run方法还没有执行完毕，也会强制终止。类似于关闭电脑时通过长按关机键的方式。

--------------------

那么如何正确的友好的中断线程呢？

下面是我们自己通过**共享变量**来实现了线程的中断，这是一种友好的方式：

public class StopThreadDemo implements Runnable {
        private static boolean stop = false; //声明共享变量，默认false
        @Override
        public void run() {
            int i = 0;
            while (!stop) { //基于共享变量来判断线程是否满足执行的条件
                System.out.println("线程在运行中..." + (++i));
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            Thread thread = new Thread(new StopThreadDemo());
            thread.start();
            Thread.sleep(500);
            stop = true; //通过共享变量赋值来中断线程运行
        }
    }

--------------------

jdk也为我们提供了**thread.interrupt()**方法来友好的中断线程

当其他线程调用了当前线程的interrupt()方法，表示向当前线程打了个招呼，告诉当前线程可以中断执行了，至于什么时候中断，取决于线程的run方法是否执行完毕。

实际上，jdk的interrupt()方法底层也是通过共享变量来中断线程，在jvm源码中有一个interrupt变量作为中断标识，Thread.currentThread().isInterrupted()方法就是获取这个变量来判断是否可以中断线程，而thread.interrupt()方法就是将变量的值从false改成true，提示线程可以中断了。

public class InterruptThreadDemo implements Runnable {
        @Override
        public void run() {
            int i = 0;
            while (!Thread.currentThread().isInterrupted()) { //判断中断标识
                System.out.println("线程在运行中..." + (++i));
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            Thread thread = new Thread(new InterruptThreadDemo());
            thread.start();
            thread.interrupt(); //提示中断线程（友好的），将标识改成ture
        }
    }

--------------------

那么什么场景需要用到interrupt()来中断线程？

线程中断的唯一依据就是线程的run方法执行结束，而while循环，wait()，join()，Thread.sleep(1000)都会阻碍到run方法的执行结束。所以在以下场景用interrupt()来中断线程是有意义的：

*  线程中有while循环
 *  线程中有thread.wait()方法阻塞线程
 *  线程中有thread.join()方法阻塞线程
 *  线程中有Thread.sleep(100)阻塞线程

当线程中有wait()，jion()，sleep(100)来阻塞线程时，如果执行interrupt()方法会唤醒线程，这时会抛出InterruptedException异常。

--------------------

Thread.interrupted()； 此方法使中断标识复位，将中断标识由true改成false。

Thread.currentThread().isInterrupted() 获取中断标识。

thread.interrupt()； 提示线程可以中断了。

###  1.5.5 start方法和run方法的区别？ ###

run方法只是调用一个实例方法，无法启动线程，而start方法可以启动一个线程

run方法可以重复调用，而start方法只能调用一次，调用多次也没有意义

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 4][]

# 2. 线程安全性分析 #

## 2.1 并发编程的源头——原子性、可见性、有序性 ##

线程安全：当多个线程同时访问某个类或对象时，每个线程访问得到都是预期的正确的结果，那么这个类就是线程安全的。

随着计算机的发展和迭代更新，发生了一下优化：

*  CPU出现了多核，增加了高速缓存，均衡与内存的速度差异 ----> 导致可见性问题
 *  操作系统增加了进程、线程、以及分时复用CPU，均衡CPU与I/O设备的速度差异 ----> 进程切换导致了原子性问题
 *  编译程序优化了指令的执行顺序（CPU层面、JVM层面指令优化重排序），使得能够更加合理的利用缓存 ----> 导致有序性问题

![2021031022201692.png][]

### 2.1.1 缓存一致性问题 ###

![20210519212710445.png][]

如果多核CPU不同核心之间的数据不一致，可以通过缓存锁或总线锁来解决一致性问题。

默认优先添加缓存锁，如果CPU不支持缓存锁，才使用总线锁。因为添加总线锁会阻塞其他CPU核心的运行，那么多核CPU的作用就失效了。

缓存锁：遵守缓存一致性协议（MESI、MOSI）

MESI表示缓存的四种状态（Modify、Exclusive、Shared、Invalidate），如果状态是Invalidate，那么让其他缓存行失效，重新从内存加载数据。

缓存行：是CPU与内存之间交互的最小单元。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 5][]

## 2.2 Java内存模型JMM ##

Java内存模型是一种抽象结构，它提供了合理的禁用缓存和禁止重排序的方法来解决可见性、有序性问题。

Java内存模型将硬件层面带来的可见性、有序性问题抽取到了Java应用层面，通过指令关键字来调取CPU提供的内存屏障或者禁用缓存指令从而解决上面的问题。

JMM提供的可见性、有序性解决方案：

*  volatile关键字：可以解决可见性、有序性问题
 *  synchronized关键字：加锁
 *  final关键字：通过final域的规则保证
 *  Happens-Before原则：前面的必须在后面的之前发生或执行

### **2.2.1 volatile** ###

> 将volatile关键字加在前，可以解决可见性、有序性问题  
> volatile底层调用了Lock指令，Lock指令有以下作用：
> 
>  *  将当前处理器缓存行的数据写入到系统内存
>  *  这个写回内存的操作会使得在其他CPU中缓存了该内存地址的数据无效
> 
> CPU层面提供了内存屏障：
> 
>  *  Store Barrier：强制所有存储屏障之前的存储指令，都在该指令屏障之前被执行，并且把存储缓冲区的数据都刷到CPU缓存
>  *  Load Barrier：强制所有load屏障之后的load指令，都在该load屏障之后被执行，并且一直等到load缓冲区被该CPU读完才能执行后面的load指令
>  *  Full Barrier：复合了Store和Load屏障的功能
> 
> Java层面也提供了内存屏障：
> 
>  *  LoadLoad Barriers：Load1 | LoadLoad | Load2  --> 确保load1在load2之前加载
>  *  StoreStore Barriers：Store1 | StoreStore | Store2 --> 确保store1在store2之前存储
>  *  LoadStore Barriers：Load1 | LoadStore | Store2 --> 确保load1加载数据在store2存储数据之前
>  *  StoreLoad Barriers：Store1 | StoreLoad | Load2 --> 确保store1存储数据并刷新到内存必须在load2加载数据之前
> 
> 本质上来说，volatile是通过内存屏障来防止指令冲排序，以及禁用CPU高速缓存来解决可见性问题。
> 
> Lock指令，本意是禁止高速缓存解决可见性问题，但实际上是一种内存屏障的功能，JMM采用Lock指令作为内存屏障来解决可见性问题。

### **2.2.2 synchronized** ###

> 修饰普通方法：锁的是当前实例对象，多个实例对象之间不影响  
> 修饰静态方法：锁的是当前类的Class对象  
> 修饰代码块：锁的是括号里配置的对象，同步代码块可以实现以上两种，并且更加灵活
> 
> jdk1.6之后对synchronized做了一下优化，减少了性能开销：
> 
>  *  自适应自旋锁
>  *  引入偏向锁、轻量级锁
>  *  锁消除、锁粗化

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 6][]

默认是偏向锁，即假设只有一个线程来抢占资源，则偏向线程A，实际开发中是不存在的，所以默认延迟开启；

如果发现有其他线程B来获取资源，发现资源已被线程A抢占，则升级锁为轻量级锁，升级锁不阻塞，会通过自旋锁的方式来判断资源是否被释放；

自旋锁是通过循环的方式来判断锁是否被释放；

如果经过几次判断后，资源依然没有被释放，则升级锁为重量级锁，等待资源释放然后再获取，重量级锁是阻塞的；

### **2.2.3 final** ###

> final与线程安全有什么关系呢？  
> 对于final域，编译器与处理器必须遵守两个重排序规则：
> 
>  *  在构造函数中对一个final修饰的变量写入，与随后把这个被构造对象的引用赋值给另一个引用变量，这两个操作之间不能重排序
>  *  初次读一个包含final域的对象的引用，与随后初次读这个final域的值，这两个操作不能重排序
> 
> 对final域写入操作的重排序规则：
> 
>  *  JMM禁止编译器把对final修饰的变量写的操作重排序到构造函数之外
>  *  编译器会在对final修饰的变量写入之后，构造函数return之前，插入StoreStore屏障，这个屏障禁止处理器把对final域的写操作重排序到构造函数之外

### **2.2.4 Happens-Before规则** ###

> Happens-Before是一种可见性规则，表示前面一个操作的结果对后续操作是可见的。
> 
> Java中有6中本身满足Happens-Before规则的规则：
> 
>  *  程序顺序规则：单线程中代码必须从前往后顺序执行
>  *  监视器锁规则：synchronized锁
>  *  volatile变量规则：对于volatile修饰的变量的写，happens-before任意后续对这个变量的读
>  *  传递性：a happens-before b, b happens-before c, 那么a happens-before c
>  *  start()规则：如果A线程执行了ThreadB.start()，那么start()操作happens-before于ThreadB中的任意其他操作
>  *  join()规则：join()方法happens-before与后面的操作，因为join()方法会阻塞主线程

### **2.2.5 原子类Automic——无锁工具** ###

和synchronized、Lock一样，可以解决原子性问题，只是Automic是无锁的，通过while循环实现。

Amomic实现原理：Unsafe类，CAS（compareAndSwapInt()方法）

AtomicInteger atomicInteger = new AtomicInteger(0);
    atomicInteger.incrementAndGet();
    atomicInteger.get();

### **2.2.6 ThreadLocal的使用和原理** ###

线程安全的一种解决方案，类似于线程的副本，多线程之间彼此隔离，互不影响.

每个线程当中都有一个ThreadLocalMap，ThreadLocalMap当中有一个默认16长度的Entry数组存储数据。

## 2.3 安全发布对象 ##

对象的发布：使一个对象能够被当前范围之外的代码所使用。

不安全发布的示例：私有属性，被其他外部调用修改

public class TestDemo {
        private String[] priAttr = {"aaa", "bbb", "ccc", "ddd"};
        public String[] getPriAttrs() {
            return priAttr;
        }
        public static void main(String[] args) throws Exception {
            TestDemo testDemo = new TestDemo();
            String[] priAttr = testDemo.getPriAttrs();
            priAttr[0] = "111"; // 结果变成了{"111", "bbb", "ccc", "ddd"};
        }
    }

**对象溢出、逃逸**

一种错误的发布，当一个对象还没有构造完成时，就使它被其他线程可见。被构造对象obj = this导致引用逃逸。

**安全发布对象的4种方法：**

*  在静态初始化函数中初始化一个对象引用
 *  通过volatile关键字
 *  添加final关键字，将对象的引用保存到构造对象的final域中
 *  通过synchronized加锁

## 2.4 J.U.C包中的工具 ##

### **2.4.1 重入锁ReentrantLock —— 独占锁** ###

一个持有锁的线程，在释放锁之前，如果再次访问了该同步锁的其他方法，不需要再次争抢锁，只需要记录重入次数。

static Lock lock = new ReentrantLock();
    public static int count = 0;
    public static void incre() {
        lock.lock(); //获得锁
        count ++; //被保护对象
        decr(); //这个时候不需要再次争抢锁
        lock.unlock(); //释放锁
    }
    public static void decr() {
        lock.lock(); //获得锁
        count --; //被保护对象
        lock.unlock(); //释放锁
    }

### **2.4.2 AQS(AbstractQueuedSynchronizer)** ###

队列同步工具

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 7][]

### **2.4.3 CountDownLatch —— 共享锁** ###

是一个同步工具类，它允许一个或多个线程一直等待，直到其他线程的操作完成之后再执行，通过倒数计数的方式来判断count=0则释放锁。

public class CountDownLatchDemo {
        public static void main(String[] args) throws InterruptedException {
            CountDownLatch countDownLatch = new CountDownLatch(3); //初始化计数count=3
            for (int i = 0; i < 3; i ++) { //那么必须要有3个线程被调用，否则将会一直等待
                new CountDownThread((i+1), countDownLatch).start();
            }
            countDownLatch.await(); //阻塞主线程，直到count计数为0
            System.out.println("主线程继续执行...");
        }
    }
    public class CountDownThread extends Thread {
        private int num;
        private CountDownLatch countDownLatch;
        public CountDownThread(int num, CountDownLatch countDownLatch) {
            this.num = num;
            this.countDownLatch = countDownLatch;
        }
        @Override
        public void run() {
            System.out.println("thread" + num + "执行...");
            countDownLatch.countDown(); //倒计时
        }
    }

从结果可以看出来，无论三个线程的顺序怎么变，但是主线程总是在三个线程全部执行完毕才会被解锁。

![20210316173352939.png][]

// 模拟高并发
    public class CountDownSimulateConcurrent implements Runnable {
        private static CountDownLatch countDownLatch = new CountDownLatch(1);
    
        public static void main(String[] args) {
            for (int i = 0; i < 1000; i ++) {
                new Thread(new CountDownSimulateConcurrent()).start();
            }
            countDownLatch.countDown(); //1000个线程全部启动后取消阻塞开始并发运行
        }
        @Override
        public void run() {
            try {
                countDownLatch.await(); //阻塞主线程
                // TODO 要测试高并发的代码
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

CountDown源码分析：CountDown有个内部类Sync继承自AQS

public CountDownLatch(int count) {//将count存入AQS的state
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count); //共享锁
    }
    Sync(int count) {
        setState(count); //state=count, 如果AQS中state=0则await被唤醒，每次调用countDown,state就会被减一
    }

### **2.4.4 Semaphore —— 共享锁** ###

信号灯，可以控制同时访问的线程的个数，通过acquire获得一个许可，如果没有许可就等待，通过release释放一个许可，类似于限流的作用。

public class SemaphoreDemo {
        public static void main(String[] args) {
            // 限制当前可以获取的最大许可数量是5
            Semaphore semaphore = new Semaphore(5);
            for (int i = 0; i < 10; i ++) {
                new Car((i + 1), semaphore).start();
            }
        }
        static class Car extends Thread {
            private int num;
            private Semaphore semaphore;
            public Car(int num, Semaphore semaphore) {
                this.num = num;
                this.semaphore = semaphore;
            }
            @Override
            public void run() {
                try {
                    semaphore.acquire(); //获得许可
                    System.out.println("第" + num + "辆车获得一个许可");
                    Thread.sleep(2000);
                    System.out.println("第" + num + "辆车释放许可");
                    semaphore.release(); //释放许可
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

Semaphore源码分析：Semaphore有个内部类Sync继承自AQS

public Semaphore(int permits) { //将许可存入AQS的state
        sync = new Semaphore.NonfairSync(permits);
    }
    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1); //获得共享锁
    }

### **2.4.5 CyclicBarrier —— 独占锁** ###

可循用使用屏障，当一组线程到达屏障点时会被阻塞，直到最后一个线程到达屏障点，所有线程的阻塞才会被打开，所有被屏障拦截的线程才会继续工作。

应用场景：当需要所有子任务都完成之后才执行主任务时，这时候可以使用CyclicBarrier。

注意：如果执行的线程数量没有达到定义的parties数量，会导致死锁，所以一般会通过两种方式解决这种问题

a. cyclicBarrier.await(10, TimeUnit.SECONDS);设置过期时间，如果到了过期时间阻塞会被打开。

b. cyclicBarrier.reset();通过reset方法重置计数，会抛出BrokenBarrierException，根据异常进行处理。

public class CyclicBarrierDemo extends Thread {
        @Override
        public void run() {
            System.out.println("主线程开始执行....");
        }
        // 定义了parties=3，那么必须是3个子线程执行完毕也会打开阻塞屏障
        // 如果少于三个子线程，那么会造成死锁
        public static void main(String[] args) {
            CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new CyclicBarrierDemo());
            new CyclicDataImportThread("path1", cyclicBarrier).start();
            new CyclicDataImportThread("path2", cyclicBarrier).start();
            new CyclicDataImportThread("path3", cyclicBarrier).start();
            System.out.println("希望三个子线程全部执行完毕之后再执行主线程后续方法....");
        }
    }
    
    public class CyclicDataImportThread extends Thread {
        private String path;
        private CyclicBarrier cyclicBarrier;
        @Override
        public void run() {
            System.out.println(path + "位置数据开始导入...");
            try {
                cyclicBarrier.await(); //阻塞
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
        public CyclicDataImportThread(String path, CyclicBarrier cyclicBarrier) {
            this.path = path;
            this.cyclicBarrier = cyclicBarrier;
        }
    }

实现原理

用到了ReentrantLock去获得锁并且释放锁，通过自循环去判断count的值是否为0，不为0则await，直到所有子线程执行完毕，count为0则breakBarrier，重置count=parties，通过signalAll()唤醒所有被阻塞的线程，并且开启新一轮的generation。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 8][]

### **2.4.6 Condition** ###

多线程协调通信的工具类，可以让某些线程一起等待某个条件condition，只有满足条件时，线程才会被唤醒。

public class ConditionDemo {
        public static void main(String[] args) {
            Lock lock = new ReentrantLock(); //使用同一个锁
            Condition condition = lock.newCondition(); //使用同一个condition
            ConditionWaitThread conditionWaitThread = new ConditionWaitThread(lock, condition);
            ConditionNotifyThread conditionNotifyThread = new ConditionNotifyThread(lock, condition);
            conditionWaitThread.start();
            conditionNotifyThread.start();
        }
    }
    public class ConditionWaitThread extends Thread {
        private Lock lock;
        private Condition condition;
        public ConditionWaitThread(Lock lock, Condition condition) {
            this.lock = lock;
            this.condition = condition;
        }
        @Override
        public void run() {
            System.out.println("begin - condition wait");
            try {
                lock.lock(); //获得锁
                condition.await(); //阻塞
                System.out.println("end - condition wait");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock(); //释放锁
            }
        }
    }
    public class ConditionNotifyThread extends Thread {
        private Lock lock;
        private Condition condition;
        public ConditionNotifyThread(Lock lock, Condition condition) {
            this.lock = lock;
            this.condition = condition;
        }
        @Override
        public void run() {
            System.out.println("begin condition notify");
            try {
                lock.lock(); //获得锁
                condition.signalAll(); //唤醒
                System.out.println("end condition notify");
            } finally {
                lock.unlock(); //释放锁
            }
        }
    }

阻塞队列中的方法

添加元素

*  add --> 如果队列满了，抛出异常
 *  offer --> true/false，添加成功返回true，否则返回false
 *  put --> 如果队列满了，则一直阻塞
 *  offer(timeout) --> 带有超时时间，如果队列满了，会阻塞timeout时长，超过阻塞时长，则返回false

移除元素

*  element --> 如果队列为空，抛出异常
 *  peek --> true/false，移除成功返回true，否则返回false
 *  take --> 如果队列为空，则一直阻塞
 *  poll(timeout) --> 如果超时了还没返回元素，则返回null

JUC包中的阻塞队列

*  ArrayBlockingQueue：基于数组结构
 *  LinkedBlockingQueue：基于链表结构
 *  PriorityBlockingQueue：基于优先级队列
 *  DelayQueue：允许延时执行的队列
 *  SynchronousQueue：没有任何存储结构的队列（在线程池Executors.newCachedThreadPool方法()里有用到）

# 3. 线程池 #

提前创建好若干个线程放在一个容器中。如果有任务需要处理，则直接将任务分配给线程池中的线程来处理，任务处理完以后当前线程不会被销毁，而是又被线程池回收，等待后续分配任务。减少线程创建、销毁的开销。

**线程使用上的问题**

*  线程的频繁创建和销毁new Thread().start()会消耗CPU内存
 *  线程的数量过多，会造成CPU资源的开销，上下文切换过于频繁也会消耗CPU资源

所以使用线程池实现线程的复用

**池化技术**

连接池、对象池、线程池、内存池。。。

池化技术的核心：复用

**线程池的好处**

*  降低创建、销毁线程带来的系统开销。
 *  提高响应速度，有新任务需要执行时不需要等待线程创建，可以立马执行。
 *  需要注意合理设置线程池大小，避免线程数过多，超过硬件资源瓶颈

**Java提供了四种线程池**

*  Executors.newFixedThreadPool(5); //创建固定数量线程的线程池
 *  Executors.newSingleThreadExecutor(); //创建只有一个线程的线程池
 *  Executors.newCachedThreadPool(); //创建一个线程数可动态伸缩的线程池，有多少请求，就创建多少个线程去处理
 *  Executors.newScheduledThreadPool(5); //创建带任务调度的线程池

public class ThreadPoolDemo implements Runnable {
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newFixedThreadPool(2);
            //ExecutorService executorService = Executors.newSingleThreadExecutor();
            //ExecutorService executorService = Executors.newCachedThreadPool();
            //ExecutorService executorService = Executors.newScheduledThreadPool(2);
            for (int i = 0; i < 10; i ++) {
                executorService.execute(new ThreadPoolDemo());
            }
            executorService.shutdown();
        }
    
        @Override
        public void run() {
            try {
                System.out.println("当前运行的线程是" + Thread.currentThread().getName());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

**线程池的基本原理**

// 四种创建线程池的方法底层都是通过此方法创建
    public ThreadPoolExecutor(int corePoolSize, //核心线程数
        int maximumPoolSize, //最大线程数
        long keepAliveTime, //超出核心线程数以外的线程空余存活时间
        TimeUnit unit, //存活时间单位
        BlockingQueue<Runnable> workQueue, //保存待执行任务的队列
        ThreadFactory threadFactory, //创建新线程的线程工厂
        RejectedExecutionHandler handler //当任务无法执行时的处理方式
    ) {}

线程池参数

*  线程数量（核心线程数量）
 *  最大线程数量
 *  阻塞队列（保存任务的队列）
 *  拒绝策略
 *  针对扩容的线程，设置存活时间，存活时间单位
 *  创建线程的工厂

**线程池执行原理**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 9][]

public void execute(Runnable command) {
        if (command == null) //如果线程为null，则报错
            throw new NullPointerException();
        int c = ctl.get(); //通过AtomicInteger获取正在工作的线程数
        // ctl是一个32位长度二进制数值，高三位代表线程状态，后29位代表线程数量
        if (workerCountOf(c) < corePoolSize) { //如果正在工作线程数 < 核心线程数
            if (addWorker(command, true)) //将创建一个new worker去执行任务
                return;
            c = ctl.get();
        }
        // workQueue.offer添加到阻塞队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // addWorker如果阻塞队列满了，添加工作线程（扩容）
        else if (!addWorker(command, false))
            reject(command); //如果添加失败，采用拒绝策略
    }

使用有界阻塞队列去存放请求执行的任务

如果请求数量非常多，队列满了，在线程池的场景中，不能去阻塞生产者线程，那么如何处理呢？

*  增加消费者线程的数量（扩容）
 *  扩容是有限制的，如果扩容解决不了问题，采用以下拒绝策略
    
     *  报错
     *  丢弃这个任务
     *  普通线程调用task.run()
     *  队列头部等待最久的任务丢弃，然后把当前任务添加到阻塞队列
     *  存储起来，等待空闲后重试（需要自定义扩展完成）

**线程池的监控**

通过自定义线程池的方式，重写线程池里面的部分方法，从而可以实现对线程池的监控。

public class CustomizedExecutors {
        public static ExecutorService newFixedThreadPool(int nThreads) {
            // 使用我们自定义的线程池
            return new CustomizedThreadPool(nThreads, nThreads,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>());
        }
    }
    
    public class CustomizedThreadPool extends ThreadPoolExecutor {
        public CustomizedThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        }
    
        @Override
        protected void beforeExecute(Thread t, Runnable r) {
            super.beforeExecute(t, r);
            // TODO 可以获取需要监控的指标和数据
        }
    
        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            super.afterExecute(r, t);
            // TODO 通过重写可以获取需要监控的指标和数据
            System.out.println("初始线程数" + this.getPoolSize());
            System.out.println("核心线程数" + this.getCorePoolSize());
            System.out.println("正在执行的任务数" + this.getActiveCount());
            System.out.println("已经完成的任务数" + this.getCompletedTaskCount());
            System.out.println("任务总数" + this.getTaskCount());
        }
    }

多线程死锁怎么办？

死锁发生的条件：

*  互斥，共享资源X和Y只能被一个线程占用。
 *  占有且等待，线程T1已经取得资源X，然后等待资源Y的时候，不释放资源X。
 *  不可抢占，其他线程不能强行抢占线程T1已经占有的资源。
 *  循环等待，线程T1等待线程T2占有的资源，线程T2等待线程T1占有的资源，循环等待。

如果解决死锁问题？

一旦发生死锁， 没什么好的办法解决，只能通过重启应用。所以要想解决死锁问题，最好的办法就是提前规避死锁。

**如何支持高并发？**

**硬件资源：**

CPU，核心数，决定了并行任务数量

内存：内存越大，那么可以存储的缓存数据就越多，响应速度就越快

磁盘：SSD，高速磁盘

网卡：千兆网卡，万兆网卡

**软件层面：**

CPU  ——> 线程，如果是8核，那么同时可以运行8个线程，通过多线程充分利用CPU资源从而提升并发数量

IO ——> 与数据库交互——> 将数据写到磁盘，通过内存/缓存/异步刷盘策略，或者数据库分库分表，分布式缓存，分布式消息中间件等技术

单节点遇到瓶颈，考虑使用多个计算机组成分布式计算机

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70]: /images/20221023/b5306d2cec8146f582746f0290fccf41.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1]: /images/20221023/d2ba382419f94b73ae0b79bd64bbfdfb.png
[20210309152340110.png]: https://img-blog.csdnimg.cn/20210309152340110.png
[20210309152405927.png]: /images/20221023/96e87335ad49441ab7af93867a631b0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 2]: /images/20221023/19798102d7c6488186d5b3622791ab92.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 3]: /images/20221023/516cc116df834bccb5de8c9ebfbd7315.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 4]: /images/20221023/9d8d89100c2841daab808f467de7b79f.png
[2021031022201692.png]: /images/20221023/74855a7a43fa4d9094b4ab8d4521e33f.png
[20210519212710445.png]: /images/20221023/012cc90108c044a594b3f231f292167d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 5]: /images/20221023/7aabe6ac19ba42d9a51c2d2d65431660.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 6]: /images/20221023/2bd3d1f394c74a7e8d8e6eba9f438bbe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 7]: /images/20221023/5ae393b1cf92434899552d390d882d08.png
[20210316173352939.png]: /images/20221023/93780d05b77b42fb99b6c7bcbb40b308.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 8]: /images/20221023/a0478220881248318435647bef859835.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 9]: /images/20221023/3202cdea283a457d9b298d9188102791.png