高并发篇_2 JUC并发编程基础和CAS原理

逃离我推掉我的手 2022-10-18 00:56 169阅读 0赞

## 高并发篇\_2 JUC并发编程基础和CAS原理 ##

**                                                                                                                                                                                                                                                                                                              田超凡**

**原创博文，仿冒必究，部分素材转载自****每特教育****蚂蚁课堂**

**1** **用户态和内核态**

**用户态和内核态都是Linux内核的概念，是运行在Linux操作系统之上的两大模块，二者之间相互联系，但是作用各不相同，最终目的是都实现了应用程序对底层硬件设备的安全访问控制。**

**用户态：****存放和运行开发者自己编写的应用程序，如自定义线程等，主要作用是调度和运行在用户空间的程序。**

**内核态：****存储和运行在操作系统之中，负责访问和协调底层硬件如磁盘、CPU实现读写和调度操作。**

1.  **用户态和内核态之间通常是请求者和调度器的角色，所有运行在用户态的用户自行编写或部署的应用程序当需要访问底层硬件设备(如访问硬盘进行数据读写，调度CPU进行上下文切换运算等)时，都需要先切换到内核态，由内核态负责调度底层硬件设备，最后再把硬件设备返回的读写结果通过网络传输的形式响应给用户态。**
2.  **用户态是不能直接访问和操作硬件设备的，这样就实现了对底层硬件设备的访问控制和安全隔离，确保硬件设备访问和操作的安全性。**
3.  **我们把用户态和内核态之间相互切换的过程称之为系统调用(System Callable)，其底层执行方式实际上是基于线程的多次阻塞和唤醒实现的。** 
4.  **Linux****使用了Ring3级别标识用户态，Ring0级别标识内核态，Ring3状态不能访问Ring0的地址空间，包括代码和数据**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70][]

**2** **线程安全同步的方式**

1.  **线程安全的定义：多线程并发运行时，保证同一时刻只有一个线程能够抢占到CPU资源得到CPU调度执行的机会，使用加锁的方式来标识抢占到资源的线程**

**线程安全同步的方式分为阻塞式和非阻塞式**

1.  **阻塞式：同一时刻只有一个线程能抢到锁，其他没有抢到锁的线程会直接变为阻塞状态(标记阻塞前执行的最后一行任务命令，随机立即从用户态切换到内核态，并一直阻塞到内核态暂停执行，直到抢到锁的线程释放锁，否则会一直阻塞)**
2.  **非阻塞式：同一时刻只有一个线程能抢到锁，其他没有抢到锁的线程不会阻塞，而是一直不断重试(不会发生系统调用，一直运行在用户态)，直到抢到锁的线程释放锁后再重新竞争。**

**3** **传统线程锁的实现方式和缺点**

**传统线程锁的实现方式：**

1.  **使用synchronized标记需要同步的方法或者结合对象锁定义同步代码块。**
2.  **使用juc并发包中的LockSupport API实现锁的阻塞和释放park()/unpark()**
3.  **使用Object类中定义的wait()和notify()阻塞和唤醒线程，需要搭配synchronized同步代码块使用。**

**传统线程锁的缺点：**

1.  **在使用synchronized1.0版本时，如果没有获取到锁的情况下则当前线程直接会变为阻塞状态----发生系统调用，从用户态切换到内核态，锁升级为重量级锁，在后期唤醒的成本会非常高，原因是每次阻塞和唤醒线程都会涉及到用户态和内核态的频繁切换，执行效率大打折扣。**

**申请锁时，从用户态进入内核态，申请到后从内核态返回用户态（两次切换）；没有申请到时阻塞睡眠在内核态。使用完资源后释放锁，从用户态进入内核态，唤醒阻塞等待锁的进程，返回用户态（又两次切换）；被唤醒进程在内核态申请到锁，返回用户态（可能其他申请锁的进程又要阻塞）。所以，使用一次锁，包括申请，持有到释放，当前进程要进行四次用户态与内核态的切换。同时，其他竞争锁的进程在这个过程中也要进行一次切换。**

1.  **重量级锁需要通过操作系统自身的互斥量（mutex lock，也称为互斥锁）来实现，然而这种实现方式需要通过用户态和核心态的切换来实现，但这个切换的过程会带来很大的性能开销，比如：Linux内核互斥锁--mutex。**

**atomic\_t count;  //****指示互斥锁的状态：**

**     1****：没有上锁，可以获得；**

**     0****：被锁定，不能获得。**

**    ** **负数：被锁定，且可能在该锁上有等待进程 ，初始化为没有上锁。**

**spinlock\_t wait\_lock;   //****等待获取互斥锁中使用的自旋锁。在获取互斥锁的过程中，操作会在自旋锁的保护中进行。初始化为锁定。**

**struct list\_head wait\_list;  //****等待互斥锁的进程队列。**

**4 CPU****上下文切换**

1.  **CPU****上下文切换**

**CPU****上下文切换只会运行在内核态中。****当多个线程同时运行时，同一时刻只有一个线程能抢占到CPU资源，CPU资源并不是固定按照每个线程的执行逻辑完整依次执行的，每个线程的CPU执行时间片不一样，抢到锁的线程只是优先被CPU开始调度，并不是说CPU先等抢到锁的线程执行完毕才继续执行下一个线程，而是不断在多个线程之间切换执行的，每次切换线程之前会记录当前线程执行到的命令行号标记为水位，在切换到其他线程执行之后，再切回当前线程执行时就会从水位开始继续往下执行。频繁的CPU上下文切换会导致内核态不堪重负进而影响线程调度和执行效率，因此实际开发中，尤其是在高并发场景之下，需要尽可能减少CPU上下文切换的频率以优化内核态性能和线程调度执行的效率。**

1.  **CPU****上下文切换的原因**

**CPU****上下文切换分为自发性上下文切换(程序代码显示调用底层SDK)和非自发性上下文切换(系统层面原因，非人为导致)**

**通过调用下列方法会导致自发性上下文切换：**

**Thread.sleep()**

**Object.wait()**

**Thread.yeild()**

**Thread.join()**

**LockSupport.park()**

**发生下列情况可能导致非自发性上下文切换：**

**切出线程的时间片用完**

**有一个比切出线程优先级更高的线程需要被运行**

**虚拟机的垃圾回收动作**

1.  **如何避免CPU上下文切换**

**无锁并发编程：****多线程竞争锁时，多线程竞争锁时，加锁、释放锁会导致比较多的上下文切换；**

**CAS****算法：****Java****的juc包中的Atomic原子类使用CAS算法来更新数据，而不需要加锁，可能会消耗cpu资源。**

**使用最少线程：****避免创建不需要的线程；**

**协程：****在单线程里实现多任务的调度，并在单线程里维持多个任务间的切换。**

**5 CAS****原理**

1.  **什么是CAS**

**CAS****，全称是Compare And Swap，翻译成中文意思是“比较并交换”，CAS是一种在内核态已经封装好，在用户态可直接调用SDK实现的线程安全的算法。**

**CAS****最大的特点是CAS是轻量级锁，只会运行在用户态，在多线程并发的场景下可以兼具提高多线程执行效率(只是运行在用户态，不会在用户态切换到内核态，更不需要阻塞和唤醒线程)和保证线程安全(通过不断重试的方式去不断获取锁，直至其他线程释放锁之后再去获取锁)，其核心算法实现是：CAS(V,E,N)**

1.  **CAS****优点和缺点**

**优点：CAS是轻量级锁，运行在用户态，在保证线程安全(不断重试获取锁)的同时不会进行系统调用(不会进行用户态和内核态之间的切换)，可以提高并发场景下多线程的调度和执行效率**

**缺点：CAS通过不断重试的方式来获取锁，是非常占用CPU资源的，长时间的空转必然会引发CPU飙高的问题，此时可以在每次重试时使用短暂线程休眠的方式来避免出现CPU飙高的问题，也可以通过设置重试次数，达到一定重试次数后再将CAS轻量级锁升级为synchronized重量级锁**

1.  **CAS****实现原理**

**CAS(V,E,N)****是实现CAS的核心算法，有3个入参：**

**V** **表示当前获取到的状态的实际值**

**E** **表示旧的预期值**

**N** **表示修改后的新值，当且仅当V=E时，将V修改为N**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70 1][]

1.  **基于CAS手写AtomicInteger原子类**

**package com.tcf.juc.exercise.day02;**

**import java.util.concurrent.atomic.AtomicInteger;**

**/\*\*\***

** \* TODO TCF** **基于CAS实现原子类AtomicInteger**

** \*/**

**public class KidAtomicInteger \{**

**    private static AtomicInteger value;**

**    // TODO TCF** **构造注入初始值**

**    public KidAtomicInteger(Integer initialValue)**

**    \{**

**        value = new AtomicInteger(initialValue.intValue());**

**    \}**

**    public KidAtomicInteger()**

**    \{**

**        value = new AtomicInteger(0);**

**    \}**

**    // TODO TCF** **获取值**

**    public Integer get()**

**    \{**

**        return value.get();**

**    \}**

**    // TODO TCF** **自增并获取最新值，基于CAS实现**

**    public Integer incrementAndGet()**

**    \{**

**        while(true)**

**        \{**

**            // TODO TCF E** **旧的预期值，每次都要重新获取，CAS不断重试**

**            int expect = value.get();**

**            if(value.compareAndSet(expect, expect+1))**

**            \{**

**                return value.get();**

**            \}**

**            // TODO TCF** **休眠，防止CPU飙高**

**            try**

**            \{**

**                Thread.sleep(50);**

**            \}**

**            catch (InterruptedException e)**

**            \{**

**                e.printStackTrace();**

**            \}**

**        \}**

**    \}**

**\}**

**编写测试类：**

**package com.tcf.juc.exercise.day02;**

**/\*\*\***

** \* TODO TCF** **测试CAS实现的原子类AtomicInteger**

** \*/**

**public class ThreadKidAtomic extends Thread \{**

**    private static KidAtomicInteger kidAtomicInteger = new KidAtomicInteger(0);**

**    @Override**

**    public void run()**

**    \{**

**        for(int i=1;i<=10000;i++)**

**        \{**

**            kidAtomicInteger.incrementAndGet();**

**        \}**

**    \}**

**    public static void main(String\[\] args)**

**    \{**

**        ThreadKidAtomic t1=new ThreadKidAtomic();**

**        ThreadKidAtomic t2=new ThreadKidAtomic();**

**        t1.start();**

**        t2.start();**

**        try**

**        \{**

**           t1.join();**

**           t2.join();**

**        \}**

**        catch (InterruptedException e)**

**        \{**

**            e.printStackTrace();**

**        \}**

**        System.out.println(kidAtomicInteger.get());**

**    \}**

**\}**

**运行结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70 2][]

1.  **基于CAS手写Lock锁**

**package com.tcf.juc.exercise.day02;**

**import java.util.concurrent.atomic.AtomicInteger;**

**/\*\*\***

** \* TODO TCF** **基于CAS无锁实现lock锁**

** \*/**

**public class KidLock2 \{**

**    // TODO TCF** **标记锁的状态，没抢到是0，抢到了是1**

**    private AtomicInteger state = new AtomicInteger(0);**

**    // TODO TCF** **上锁**

**    public void lock()**

**    \{**

**        while(true)**

**        \{**

**            // TODO TCF** **获取旧的预期值E**

**            int expect = state.get();**

**            if(expect==0 && state.compareAndSet(expect,1))**

**            \{**

**                return;**

**            \}**

**            // TODO TCF** **线程休眠，防止CPU飙高**

**            try**

**            \{**

**                Thread.sleep(50);**

**            \}**

**            catch (InterruptedException e)**

**            \{**

**                e.printStackTrace();**

**            \}**

**        \}**

**    \}**

**    // TODO TCF** **释放锁**

**    public void unlock()**

**    \{**

**        while(true)**

**        \{**

**            // TODO TCF** **获取旧的预期值E**

**            int expect = state.get();**

**            if(expect == 1 && state.compareAndSet(expect,0))**

**            \{**

**                return;**

**            \}**

**            // TODO TCF** **线程休眠，防止CPU飙高**

**            try**

**            \{**

**                Thread.sleep(50);**

**            \}**

**            catch (InterruptedException e)**

**            \{**

**                e.printStackTrace();**

**            \}**

**        \}**

**    \}**

**\}**

**编写测试类：**

**package com.tcf.juc.exercise.day02;**

**/\*\*\***

** \* TODO TCF** **基于CAS实现Lock锁多线程测试**

** \*/**

**public class ThreadKidLock2 extends Thread \{**

**    private static int sum=0;**

**    private static KidLock2 lock = new KidLock2();**

**    @Override**

**    public void run()**

**    \{**

**        lock.lock();**

**        for(int i=1;i<=30000;i++)**

**        \{**

**            sum++;**

**        \}**

**        lock.unlock();**

**    \}**

**    public static void main(String\[\] args)**

**    \{**

**        ThreadKidLock2 t1 = new ThreadKidLock2();**

**        ThreadKidLock2 t2 = new ThreadKidLock2();**

**        t1.start();**

**        t2.start();**

**        try**

**        \{**

**            t1.join();**

**            t2.join();**

**        \}**

**        catch (InterruptedException e)**

**        \{**

**            e.printStackTrace();**

**        \}**

**        System.out.println(sum);**

**    \}**

**\}**

**运行结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70 3][]

1.  **CAS****的ABA问题及其解决方案**

**ABA****问题是基于JUC并发包中的原子类Atomic实现CAS的一个BUG，问题现象描述:**

**假设Atomic初始状态是0(V=0)，有3个线程并行运行，执行步骤如下：**

**Step1  Thread-0  ****设置N=1 -> V=0，E=0，V=E，修改成功 -> 此时V=1**

**Step2  Thread-1  ****设置N=0 -> V=1，E=0，V!=E，修改失败 -> 此时V=1**

**Step3  Thread-2 ** **设置N=1 -> V=1，E=1，V=E，修改成功 -> 此时V=2**

**这样就发生了T0和T2线程都同时修改V成功了的情况，导致线程不安全问题发生，这一类问题就是使用上文基于原子类手写CAS实现轻量级锁的BUG。那么这类问题怎么解决呢？**

**上文提到，多个线程并发执行获取CAS锁的时候，当且仅当旧的预期值E=实际值V，才会获取到锁并将实际值V修改为目标值N，考虑到ABA的问题，仅使用一个条件作为获取锁的判定依据显然是不充分的，此时可以再加入一个条件，通过版本号来控制。(当且仅当E=V并且 当前线程操作的版本号=旧的预期值版本号时，当前线程才能获取到CAS锁)**

**原子类实现CAS的ABA问题解决方案：**

**使用juc包下的AtomicStampReference实现同时根据E==V和版本号对比两个条件来获取CAS轻量级锁，解决ABA问题代码如下：**

**package com.tcf.juc.exercise.day03;**

**import lombok.extern.slf4j.Slf4j;**

**import java.util.concurrent.atomic.AtomicStampedReference;**

**/\*\*\***

** \* TODO TCF** **基于AtomicStampReference记录版本号解决CAS ABA问题**

** \*/**

**public class ThreadCasABA extends Thread \{**

**    // TODO TCF** **原子类实现CAS锁**

**    private static AtomicStampedReference<Integer> casStatus = new AtomicStampedReference<Integer>(0,0);**

**    // TODO TCF** **最大更新次数**

**    public static int UPDATE\_NUM = 200;**

**    @Override**

**    public void run()**

**    \{**

**        // TODO TCF** **旧的预期值E**

**        int except = casStatus.getReference();**

**        System.out.println("****当前线程：" + Thread.currentThread() + " 旧的预期值：" + except);**

**        // TODO TCF** **当前线程版本号**

**        int version = casStatus.getStamp();**

**        try**

**        \{**

**            Thread.sleep(1000);**

**        \}**

**        catch (InterruptedException e)**

**        \{**

**            e.printStackTrace();**

**        \}**

**        // TODO TCF** **基于CAS实现轻量级锁，使用版本号控制，防止出现ABA问题**

**        if(casStatus.compareAndSet(except, except+1, version, version+1))**

**        \{**

**            System.out.println("****当前线程：" + Thread.currentThread().getName() + "\\t当前值为：" + casStatus.getReference() + "\\t版本号为：" + casStatus.getStamp());**

**        \}**

**        else**

**        \{**

**            System.out.println("****当前线程：" + Thread.currentThread().getName() + "\\t版本号不同，更新失败");**

**        \}**

**    \}**

**    public static void main(String\[\] args)**

**    \{**

**        ThreadCasABA t1 = new ThreadCasABA();**

**        ThreadCasABA t2 = new ThreadCasABA();**

**        ThreadCasABA t3 = new ThreadCasABA();**

**        t1.start();**

**        t2.start();**

**        t3.start();**

**        try**

**        \{**

**            t1.join();**

**            t2.join();**

**            t3.join();**

**        \}**

**        catch (InterruptedException e)**

**        \{**

**            e.printStackTrace();**

**        \}**

**    \}**

**\}**

**执行结果如下：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70 4][]

**附：**

**基于原子类实现CAS可解决ABA问题的核心API方法说明：**

**AtomicStampedReference<V>**

**public boolean compareAndSet(V   expectedReference,**

**                                 V   newReference,**

**                                 int expectedStamp,**

**                                 int newStamp)**

**参数说明：**

**（1）第一个参数expectedReference：表示预期值。**

**（2）第二个参数newReference：表示要更新的值。**

**（3）第三个参数expectedStamp：表示预期的时间戳。(旧的预期的版本号)**

**（4）第四个参数newStamp：表示要更新的时间戳。(更新后的版本号)**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70]: /images/20221014/fc8a43e1bb8b4146ba7235896479c266.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70 1]: /images/20221014/aa81e948d962403a9d8523676ce028d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70 2]: /images/20221014/86d01ff0bb664ef99916f105df83de80.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70 3]: /images/20221014/da9b8a0bbeac463f90b1a501664c10eb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwMDU2MzQx_size_16_color_FFFFFF_t_70 4]: /images/20221014/e35c636720dc4f7eb690f88e966b7eb6.png