JAVA并发编程总结

Posted 2023-02-06 我永远信仰

写在前面

​ 本篇内容是学习记录的一些笔记，在学习过程中有许多疑惑，通过多写Demo测试验证自己的想法，该过程比较杂乱无章，所以本篇内容更侧重于记录结论和自己的一些总结以及一些辅助自己容易回忆起的简单Demo等等。但也就如此而已，如果我真想不起来，还是更愿意去找网上的文章看看，毕竟人家的文章足够的好。
​ 即使每个知识点网上都有，但我觉得通过自己整理过的东西会更容易理解，同时也能加深自己的记忆，而且在整理过程中，自己脑袋偶尔会产生一些想法，“如果我这么做，那他得到的结果会不会还是一样呢”，我不会放过它，这说不定能让我更深一步领悟该知识点。

学习收获

• api的使用（也就是JDK提供的各个并发类的玩法）
• api的了解，知道了可以拥有各种骚操作的玩法
  • 如果遇到这方面的问题或者情景，知道有什么东西可以使用，提供给我问题排查的思路
• 学习到了一些思维，比如：无锁–CAS、AQS的原理—以及里面的等待队列唤醒机制、不会让CPU空转浪费资源、公平锁与非公平锁。
• 同步工具类底层实现都是unsafe类，unsafe类的使用
• 源码级别–shutdown和shutdownNow的区别，抛出中断异常底层是因为shutdownNow调用了interrupt方法
• 一些并发模式的思想
• 大任务分治处理，无序–Future 、有序–ForkJoin
• ……

文章目录

• 写在前面
• 学习收获
• 1. 并发编程基础篇
• • 线程中断(interrupt)
  • 线程安全概念
  • Synchronized概念
  • 类锁和对象锁
  • 对象锁的同步和异步
  • 并发脏读问题
  • synchronized锁重入
  • 抛出异常和锁的关系
  • Synchronized代码块和锁失效问题
  • 线程之间通讯
  • 守护线程和用户线程
• 2. 并发编程进阶篇
• • volatile关键字
  • volatile和static的区别
  • Atomic类的原子性
  • CAS原理解析
  • ThreaLocal和InheritableThreadLocal
  • Unsafe类安全限定
  • Unsafe实操
  • Unsafe线程调度
  • 同步并发类容器
  • 常见并发队列
• 3. 并发编程精通篇
• • CountDownLatch工具类
  • CyclicBarrier工具类
  • Phaser（阶段器/移相器）
  • Semaphore工具类
  • Exchanger交换器
  • ReentrantLock和ReentrantReadWriteLock
  • StampedLock简介&写锁writeLock
  • LockSupport
  • AbstractQueuedSynchronizer（AQS）
  • 锁的分类
  • • 乐观锁和悲观锁
    • 公平锁与非公平锁
    • 独占锁与共享锁
    • 可重入锁
    • 自旋锁
  • ThreadLocalRandom与Random类
  • LongAdder高性能累加器
  • Accumulator高性能累加器
  • [
  • 线程池拒绝策略
  • ThreadFactory
  • 处理线程池内未捕获异常
  • 线程池shutdown 和shutdownNow方法
  • 线程池状态
  • 核心线程超时策略
  • 线程池其他用法
  • 线程及线程池切面
  • 移除线程池中的任务
  • Future模式
  • Master-Worker模式
  • CompletionService
  • ForkJoin

1. 并发编程基础篇

线程中断(interrupt)

• 线程调用interrupt()方法不会让线程中断，只会给线程设置一个中断的标志（设置中断的flag为true），具体的中断仍需我们自己写程序来控制

其他方法：

• Thread.currentThread().isInterrupted():获取当前线程的中断标记，调用此方法不会改变中断的状态，也就是说这个只是简单的get操作
• Thread.interrupted() ：不同于上面的方法，这是一个静态方法， 也是获取当前线程的中断标记，但是调用此方法会改变中断的状态，清除了线程的中断标记。也就是说get操作后会将是否处于中断的标记设置为false。

public boolean isInterrupted() 
    return isInterrupted(false);


public static boolean interrupted() 
    return currentThread().isInterrupted(true);


//demo
public class InterruptsTest 
    public static void main(String[] args) throws InterruptedException 
        Thread thread = new Thread(() -> 
            for (int i = 0; i <= 99999; i++) 
                if (Thread.interrupted()) 
                if (Thread.currentThread().isInterrupted()) 
                    System.out.println(Thread.currentThread().getName() + i + "线程处于中断状态");
                    System.out.println(Thread.currentThread().isInterrupted());
                    break;
                
                System.out.println(Thread.currentThread().isInterrupted());
            
        );
        thread.start();
        // 先睡一会觉
        Thread.sleep(20);
        // 再中断,
        thread.interrupt();
        System.out.println("main程序");
    

如果该线程在调用 Object 类的 wait()方法或 join()时被阻塞, sleep(long), 这个类的方法，那么它的中断状态会被清除并且会收到一个InterruptedException（中断异常）

如果遇到了中断异常，我们不必恐惧它，它的出现 说明我们在调用中断的时候被中断的线程中执行着sleep()、wait()、join()方法，可以利用这个异常来进行一些数据补偿之类的操作。

线程安全概念

1. 线程安全的概念:当多个线程访问某一个类、对象或方法时，这个类、对象或方法都能表现出与单线程执行时一致的行为，那么这个类、对象或方法就是线程安全的。
2. 线程安全问题都是由全局变量及静态变量引起的。
3. 若每个线程中对全局变量、静态变量只有读操作，而无写操作，一般来说，这个全局变量是线程安全的;若有多个线程同的执行写操作，一般都需要考虑线程同步，否则的话就可能影响线程安全。

Synchronized概念

• Synchronized的作用是加锁，所有的synchronized方法都会顺序执行，(这里只占用CPU的顺序）。
• Synchronized方法执行方式:
  • 首先尝试获得锁
  • 如果获得锁，则执行Synchronized的方法体内容。
  • 如果无法获得锁则等待，并且不断的尝试去获得锁，一旦锁被释放，则多个线程会同时去尝试获得锁，造成锁竞争问题。

锁竞争问题，在高并发、线程数量高时会引起CPU占用居高不下，或者直接宕机。

类锁和对象锁

• Synchronized作用在非静态方法上代表的对象锁（理解为资源），一个对象一个锁、多个对象之间不会发生锁竞争（各自获取自己的资源，就不会发生竞争）
• Synchronized作用在静态方法上则升级为类锁，所有对象共享一把锁，存在锁竞争（静态是供大家一起使用的，所以会发生锁竞争）。

对象锁的同步和异步

• 对象锁只针对synchronized修饰的方法生效、对象中的所有synchronized方法都会同步执行、而非synchronized方法异步执行
• 避免误区:类中有两个synchronized方法，两个线程分别调用两个方法，相互之间也需要竞争锁，因为两个方法属于同一个对象，而我们是在对象上加锁

并发脏读问题

• 多个线程访问同一个资源，在一个线程修改数据的过程中，有另外的线程来读取数据，就会引起脏读的产生。
• 为了避免脏读我们一定要保证数据修改操作的原子性、并且对读取操作也要进行同步控制

银行存钱和取钱的例子

synchronized锁重入

• 同一个线程得到了一个对象的锁之后，再次请求此对象时可以再次获得该对象的锁。
• 同一个对象内的多个synchronized方法可以锁重入
• 父子类可以锁重入，不用担心死锁问题

理解：自己拿到了资源自己可以反复的利用

抛出异常和锁的关系

一个线程在获得锁之后执行操作，发生错误抛出异常，则自动释放锁

1. 可以利用抛出异常，主动释放锁
2. 程序异常时防止资源被死锁、无法释放
3. 异常释放锁可能导致数据不一致

Synchronized代码块和锁失效问题

Synchronized代码块:

• 同类型锁之间互斥,不同类型的锁之间互不干扰(比如类锁和对象锁互不影响)

不要在线程中修改对象锁的引用，引用被改变会导致锁失效

• 在线程中修改了锁对象的属性,而不修改引用则不会引起锁失效、不会产生线程安全问题。如代码//1
• 线程A修改了对象锁的引用，则线程B实际的到了新的对象锁，而不是锁被释放了,因此引发了线程安全问题。如代码//2

Person person = new Person();

synchronized(person)
    //1
    person.setName("周三");

    //2
    person = new Person();

线程之间通讯

• 每个线程都是独立运行的个体，线程通讯能让多个线程之间协同工作
• Object类中的wait/notify方法可以实现线程间通讯
• Wait/notify必须与synchronized一同使用
• Wait释放锁、notify不释放锁

使用关键字volatile定义的变量也可以实现同样的效果

• Notify只会通知一个wait中的线程，并把锁给他，不会产生锁竞争问题，但是该线程处理完毕之后必须再次notify或notifyAll，完成类似链式的操作。
• NotifyAll会通知所有wait中的线程，会产生锁竞争问题。（理解:当同时通知所有等待的线程，这些线程争抢同一个资源）

守护线程和用户线程

• 线程分类: daemon线程（守护线程)、user线程(用户线程)
• 易混淆知识点:main函数所在的线程就是一个用户线程
• 重要知识点1∶最后一个user线程结束时，JVM会正常退出（在main启动时会启动一个DestroyJVM的用户线程来关闭JVM），不管是否有守护线程正在运行。反过来说∶只要有一个用户线程还没结束，JVM进程就不会结束。
• 重要知识点2∶父线程结束后，子线程还可以继续存活，子线程的生命周期不受父线程的影响

2. 并发编程进阶篇

volatile关键字

作用在类的变量上

• 强制线程到共享内存中读取数据，而不从线程工作内存中读取，从而使变量在多个线程间可见。
• volatile无法保证原子性，volatile属于轻量级的同步，性能比synchronized强很多(因为不加锁)，但是只保证线程见的可见性，并不能替代synchronized的同步功能。netty框架中大量使用volatile

volatile只能解决可见性，不能保证原子性

volatile和static的区别

• Static保证唯一性,不保证一致性，多个实例共享一个静态变量。
• Volatile保证一致性，不保证唯一性，多个实例有多个volatile变量

static变量的修改方式和普通变量一样都是先拷贝再修改后写回主内存，只有一份（唯一性）
volatile直接在主内存中修改（可见性、一致性），每个实例都有单独的变量，只是一个普通变量

Atomic类的原子性

• 使用Atomiclnteger等原子类可以保证共享变量的原子性
• 但是Atomic类不能保证成员方法的原子性

原理是使用CAS

CAS原理解析

1. JDK提供的非阻塞原子操作，通过硬件保证了比较、更新操作的原子性
2. JDK的Unsafe类提供了一系列的compareAndSwap*方法来支持CAS操作

CAS-ABA问题：解决->（增加版本号、使用时间戳）

ThreaLocal和InheritableThreadLocal

使用ThreadLocal维护变量时，ThreadLocal为每个使用该变量的线程提供独立的变量副本，所以每一个线程都可以独立地改变自己的副本，而不会影响其它线程所对应的副本

• 父线程使用ThreadLocal创建的变量，在子线程中获取不到
• 如果子线程想要得到可以使用InheritableThreadLocal，且子线程之间不会互相影响

使用InheritableThreadLocal的

• 好处是可以知道他的父线程，这样就方便我们跟踪程序执行到了什么地方
• 与ThreadLocal不同的是local变量的初始化时机（在Thread类里面找），在线程创建的时候会判断父线程是否是InheritableThreadLocal，如果是的话就将父线程的值赋值给当前线程，否则赋值为null，所以他的get方法会返回父线程的值

小结：

• Thread类中的threadLocals、inheritableThreadLocals成员变量为ThreadLocal.ThreadLocalMap对象Map的key值是ThreadLocal对象本身，查看set、set、remove方法
• Thread无法解决继承问题，而InheritableThreadLocal可以
• InheritableThreadLocal继承自ThreadLocal
• InheritableThreadLocal可以帮助我们做链路追踪、日志输出等

Unsafe类安全限定

AtomicXXX类的底层就是依赖Unsafe类里的相关方法（本地方法）来完成原子性操作

• Unsafe类是单例的
• 使用Unsafe类可以直接操控内存，所以他是不安全的类
• Unsafe类在rt.jar包里，是核心类，是BootStrapClassLoad来加载的，我们不能调用他的方法，原因是他会判断是否是根加载器调加载的类调用的他方法，如果不是抛出异常，所以Atomic类能调用
• 需要看它的源码，可以去github

@CallerSensitive
    public static Unsafe getUnsafe() 
        Class var0 = Reflection.getCallerClass();
        if (!VM.isSystemDomainLoader(var0.getClassLoader())) 
            throw new SecurityException("Unsafe");
         else 
            return theUnsafe;
        
    

Unsafe实操

Unsafe -突破安全限制
通过反射模式可以突破Unsafe类的安全限制

putInt、getInt、getAndSetInt、getAndAddInt、CAS、实例demo:

public class UnsafeDemo01 

    private int age;

    public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException 
        UnsafeDemo01 demo01 = new UnsafeDemo01();
        // 通过反射获取Unsafe的静态变量成员
        Field field = Unsafe.class.getDeclaredField("theUnsafe");
        // 设置无障碍
        field.setAccessible(true);
        // 强转获取Unsafe对象
        Unsafe unsafe = (Unsafe) field.get(null);
        // 获取age属性的内存偏移地址
        long ageOffset = unsafe.objectFieldOffset(UnsafeDemo01.class.getDeclaredField("age"));
        // 设置age值为11
        unsafe.putInt(demo01, ageOffset, 11);
        System.out.println("unsafe的put方法设置age ：" + demo01.age);
        // 获取age
        int age = unsafe.getInt(demo01, ageOffset);
        System.out.println("unsafe的get方法获取age : " + age);
        // 验证CAS方法,如果它是10,设置为88,cas不一定保证成功,有默认的自旋次数,失败
        boolean flag = unsafe.compareAndSwapInt(demo01, ageOffset, 10, 88);
        System.out.println("CAS test ：" + flag + " value is " + demo01.age);
        // 成功
        flag = unsafe.compareAndSwapInt(demo01, ageOffset, 11, 666);
        System.out.println("CAS test ：" + flag + " value is " + demo01.age);
        // 获取并设置,返回的是旧的值
        int setInt = unsafe.getAndSetInt(demo01, ageOffset, 333);
        System.out.println("old value :" + setInt + "new value :" + demo01.age);
        // 获取并增加
        int addInt = unsafe.getAndAddInt(demo01, ageOffset, 7);
        System.out.println("old value :" + setInt + "new value :" + demo01.age);
    

• 如果是针对static变量操作，则上面的方法不是再操作实例，而是需要操作类模板，例如

private static int age;

...

unsafe.staticFieldOffset(UnsafeDemo01.class.getDeclaredField("age"));

int age = unsafe.getInt(UnsafeDemo01.class, ageOffset);

• 如果是针对volatile变量操作，则几乎和普通变量没区别，只是put、get方法调用改成有volatile后缀的方法

• 如果是数组，则传入数组类模板来获取地址偏移量，设置数组元素值 使用偏移量+类型大小*下标

Unsafe unsafe = (Unsafe) field.get(null);
long arrOffset = unsafe.arrayBaseOffset(long[].class);

// 设置数组下标为1的元素的值为9
unsafe.putLong(demo01.getArray(),arrOffset + arrSize * 1)

**注意：**如果设置的大小超过了数组的长度，不会报错，但也不会扩容

• 直接操作内存**（危险性操作体现）**

public native long allocateMemory(long bytes);//分配内存
public native long reallocateMemory(long address, long bytes);//重新分配内存
public native void setMemory(Object o, long offset, long bytes, byte value);//初始化内存
public void setMemory(long address, long bytes, byte value)//初始化内存
public native void copyMemory(Object srcBase, long srcOffset, Object destBase, long destOffset, longbytes);//复制内存
public void copyMemory(long srcAddress, long destAddress, long bytes);//复制内存
public native void freeMemory(long address);//释放内存

有C++基础理解起来不难

Unsafe线程调度

1. public native void park(boolean isAbsolute, long time);挂起线程
2. public native void unpark(Object thread);唤醒线程
3. 需要注意线程的interrupt方法同样能唤醒线程，但是不报错
4. java.util.concurrent.locks.LockSupport使用unsafe实现，可以点进去看一下该类的源码和相关方法

public class UnsafeDemo02 
    public static void main(String[] args) throws Exception 
        UnsafeDemo01 demo01 = new UnsafeDemo01();
        // 通过反射获取Unsafe的静态变量成员
        Field field = Unsafe.class.getDeclaredField("theUnsafe");
        // 设置无障碍
        field.setAccessible(true);
        // 强转获取Unsafe对象
        Unsafe unsafe = (Unsafe) field.get(null);

        Thread thread = new Thread(() -> 
            System.out.println(Thread.currentThread().getName() + "--start");
            // 线程挂起2000 * 100000 纳秒 = 2s
            unsafe.park(false, 2000 * 1000000);
            if (Thread.currentThread().isInterrupted()) 
                System.out.println("Thread status is interrupted");
            
            // 挂起当前时间往后推2s
            //unsafe.park(true, System.currentTimeMillis() + 2000);
            System.out.println(Thread.currentThread().getName() + "--end");
        );
        thread.start();
        // 唤醒线程
        unsafe.unpark(thread);
        // 设置中断flag
        //thread.interrupt();
        System.out.println("main run over");
    

同步并发类容器

• Collections.synchronizedXXX()：可以将集合变成线程安全的同步类容器，原理是在整个集合上加了synchronized锁，并发会比较差
• jdk1.5后增加了Concurrent并发类容器：
  • ConcurrentHashMap替代HashMap、HashTable
  • ConcurrentSkipListMap替代TreeMap
  • ConcurrentHashMap：将hash表分为16个segment（分区），每个segment单独进行锁控制，从而减小了锁的粒度，提升了性能。
• COW(copy on write)

COWlterator的弱一致性

• 使用COW容器的iterator方法实际返回的是COWlterator实例，遍历的数据为快照数据，其他线程对于容器元素增加、删除、修改不对快照产生影响。
• 对java.util.concurrent.CopyOnWriteArrayList、java.util.concurrent.CopyOnWriteArraySet均适用。

常见并发队列

并发-无阻塞队列

• **ConcurrentLinkedQueue**：无阻塞、无锁、高性能、无界、线程安全，性能优于BlockingQueue（并发阻塞）、不允许null值

并发-阻塞队列

• **ArrayBlockingQueue**:基于数组实现的阻塞有界队列、创建时可指定长度，内部实现维护了一个定长数组用于缓存数据,内部没有采用读写分离，写入和读取数据不能同时进行，不允许null值

使用的时候要注意，该队列的offer、add、put方法有很大区别

• **LinkedBlockingQueue** :基于链表的阻塞队列,内部维护一个链表存储缓存数据，支持写入和读取的并发操作，创建时可指定长度也可以不指定，不指定时代表无界队列，不允许null值
• **SynchronousQueue **:没有任何容量，必须先有线程先从队列中take,才能向queue中add数据，否则会抛出队列已满的异常。不能使用peek方法取数据,此方法底层没有实现,会直接返回null
• **PriorityBlockingQueue**:一个无界阻塞队列，默认初始化长度11，也可以手动指定，但是队列会自动扩容。资源被耗尽时导致OutOfMemoryError。不允许使用null元素。不允许插入不可比较的对象（导致抛出 ClassCastException）,加入的对象实现comparable接口（维护了take取出元素的优先级）
• **DelayQueue**：https://www.cnblogs.com/myseries/p/10944211.html

3. 并发编程精通篇

CountDownLatch工具类

**CountDownLatch**是一个辅助工具类（可以理解为一个同步计数器），它允许一个或多个线程等待系列指定操作的完成。CountDownLatch 以一个给定的数量初始化。countDown()每被调用一次，这一数量就减一。通过调用await()方法，线程可以阻塞等待这一数量到达零（当计数器数值减为0时，所有受其影响而等待的线程将会被激活）。

• 只能使用一次，计数器的值只能在构造方法中初始化一次，之后无法再改变它的值

CyclicBarrier工具类

它的作用就是会让所有线程都等待完成后才会继续下一步行动

CyclicBarrier 使用场景
可以用于多线程计算数据，最后合并计算结果的场景。（CountDownLatch也可以）
CyclicBarrier 与 CountDownLatch 区别

• CountDownLatch 是一次性的，CyclicBarrier 是可循环利用的
• CountDownLatch 参与的线程的职责是不一样的，有的在倒计时，有的在等待倒计时结束。CyclicBarrier 参与的线程职责是一样的。

public class CyclicBarrierDemo 
    public static void main(String[] args) 
        test1();
    

    private static void test1() 
        int size = 4;
        ExecutorService executorService = Executors.newFixedThreadPool(size);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(size, () -> 
            System.out.println("最后一名是：" + Thread.currentThread().getName());
        );
        for (int i = 0; i < size; i++) 
            Runnable r = () -> 
                try 
                    Thread.sleep(1000);
                    System.out.println(Thread.currentThread().getName() + " :达到栅栏点A");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread().getName() + " :从栅栏点A出发");

                    Thread.sleep(2000);
                    System.out.println(Thread.currentThread().getName() + " :达到栅栏点B");
                    cyclicBarrier.await();
                    System.out.println
以上是关于JAVA并发编程总结的主要内容，如果未能解决你的问题，请参考以下文章
 
 Java并发总结-全景图
 并发编程：我对Java并发编程的总结和思考
 JAVA并发编程学习总结
 并发编程总结——java线程基础1
 关于Java并发编程的总结和思考
 Java并发编程小总结：CountDownLatchCyclicBarrier和Semaphore