【Java核心知识】JUC包相关知识

ゝ一纸荒年。 2024-03-09 00:42 138阅读 0赞

#### 文章目录 ####

*  JUC包主要内容
 *  Java内置锁
 *   *  为什么会有线程安全问题
     *  Synchronize锁
     *  Java对象结构
     *  Synchronize锁优化
     *  线程间通信
     *  Synchronize与wait原理
 *  CAS和JUC原子类
 *   *  CAS原理
     *  \`JUC\`原子类
     *  \`ABA\`问题
 *  可见性和有序性
 *   *  为什么会有可见性
     *  参考链接
 *  显式锁
 *   *  Lock接口常用方法
     *  显式锁分类
     *  显式锁实现原理
     *  参考链接

## JUC包主要内容 ##

JUC包是与并发编程相关的包，主要包含四部分`锁`、`原子类`、`并发集合`、`多线程`，如下图所示。

其中，

*  锁可以分为内置锁和显式锁；
 *  原子类主要是一些通过`CAS`实现的原子类；
 *  并发集合主要就是一些线程安全的集合，比如`ConcurrentHashMap`,`BlockQueue`等；
 *  多线程包括callable接口和线程池等；

![在这里插入图片描述][650d444dd7914fdeb8db42d035d3a4eb.png]

## Java内置锁 ##

### 为什么会有线程安全问题 ###

`i++`线程不安全的原因在于自增操作不是`原子性`的，可以分为三步：`内存取值`、`寄存器加1`、`存值到内存`。

除了原子性之外，`可见性`和`有序性`也会导致线程安全问题。可见性是指线程B并不一定能够及时看到线程A对变量的修改。

### Synchronize锁 ###

Synchronize关键字可以`作用在方法上`，也可以`作用于代码块上`，本质上都是`锁住了某个对象`，但synchronize作用于方法上是一种粗粒度的锁，会导致其他线程也不能访问该对象的其他方法。

在`JVM`的堆中，每个对象都会有一个对象监视器，synchronize就是锁住了这个`对象监视器`，从而保证了原子性。

那么如何保证`可见性`呢？线程加锁时，必须清空工作内存中共享变量的值，从而使用共享变量时需要从主内存重新读取；线程在解锁时，需要把工作内存中最新的共享变量的值写入到主存，以此来保证共享变量的可见性。(这里是个泛指，不是说只有在退出synchronized时才同步变量到主存)

### Java对象结构 ###

Java的对象都放在JVM的堆中，每个对象的结构包括：

*  对象头：
    
     *  Mark Word:记录哈希码，GC标志位、锁状态等信息。不同锁状态下Mark Word是不同的，但最后两位都代表了锁状态。
     *  类对象指针：指向方法区的该类相关信息
     *  数组长度：如果对象是数组才有此结构
 *  对象体：包含对象的实例变量，包含父类的实例变量
 *  对齐字节：为了保证8字节的对齐而填充的数据

### Synchronize锁优化 ###

为了优化`Synchronize`锁的性能，Java提出了逐步升级的四种锁：无锁->偏向锁->轻量级锁->重量级锁。

*  无锁：
 *  偏向锁：Mark Word中存储持有锁的线程ID，当有线程执行时，先判断对象头的线程ID是否与此线程ID相等，如果相等，直接向下执行；如果不相等，说明存在竞争，锁升级为轻量级锁。
 *  轻量级锁：对象头存储持有锁的线程ID，将对象头原来的哈希码放入线程栈帧中的锁记录中。当别的线程竞争锁时，不会立即阻塞，切换用户态，而是会自旋，然后使用`CAS`尝试获取锁，降低了阻塞线程的消耗。自旋等待时间和上一个竞争线程等待结果有关：如果上一个竞争线程自旋成功了，那么这次自旋的次数会更多；如果上一个竞争线程自旋失败了，那么这次自旋的次数会减少。自旋不会一直持续下去，如果超过了指定时间，会膨胀为重量级锁！
 *  重量级锁：重量级锁对象头会指向一个监视器对象（每个对象都有一个监视器对象），该`监视器`通过三个队列（竞争队列、阻塞队列、等待时间片的就绪队列）来登记和管理排队的线程，会涉及到线程的阻塞，切换用户态。

**轻量级锁执行过程：**

*  1、判断对象是否加锁，如果没加锁，进行以下操作
 *  2、在自己的栈帧中创建锁记录，用来存放加锁对象的哈希码
 *  3、创建好锁记录后，通过CAS自旋操作，尝试将锁对象头的锁记录指针替换成栈帧中锁记录的地址
 *  4、替换栈帧后会返回锁对象的哈希码，然后填入栈帧的锁记录中

### 线程间通信 ###

可以使用`Object`的`wait()`,`notify()`方法来进行线程间的通信。

**wait()方法的原理**：

1）当线程调用了`lock`（某个同步锁对象）的`wait()`方法后，`JVM`会将当前线程加入`lock`监视器的`WaitSet`（等待集），等待被其他线程唤醒。  
2）当前线程会释放`lock`对象监视器的`Owner`权利，让其他线程可以抢夺`lock`对象的监视器。  
3）让当前线程等待，其状态变成`WAITING`。

**notify()方法的原理**：

1）当线程调用了`lock`（某个同步锁对象）的`notify()`方法后，`JVM`会唤醒`lock`监视器`WaitSet`中的第一个等待线程。  
2）当线程调用了`lock`的`notifyAll()`方法后，`JVM`会唤醒`lock`监视器`WaitSet`中的所有等待线程。  
3）等待线程被唤醒后，会从监视器的`WaitSet`移动到`EntryList`，线程具备了排队抢夺监视器`Owner`权利的资格，其状态从`WAITING`变成`BLOCKED`。  
4）`EntryList`中的线程抢夺到监视器`Owner`权利之后，线程的状态从`BLOCKED`变成`Runnable`，具备重新执行的资格。

**缓冲队列**

/**
     * 生产者消费者队列
     */
    //数据缓冲区，类定义
    public class DataBuffer<T> {
        
        public static final int MAX_AMOUNT = 10; //数据缓冲区最大长度
        //保存数据
        private List<T> dataList = new LinkedList<>();
        //数据缓冲区长度
        private Integer amount = 0;
        // 用来保证只有一个线程存元素或者取元素
        private final Object LOCK_OBJECT = new Object();
        // 当队列满了后，用于阻塞生产者
        private final Object NOT_FULL = new Object();
        // 当队列为空时，用于阻塞消费者
        private final Object NOT_EMPTY = new Object();
        // 向数据区增加一个元素
        public void add(T element) throws Exception
        {
        
            // 队列已满，不能存元素
            while (amount > MAX_AMOUNT)
            {
        
                synchronized (NOT_FULL)
                {
        
                    System.out.println("队列已经满了！");
                    // 等待未满通知，这里为什么需要wait，是因为需要等待一个条件满足，而不能只用synchronize，某一时刻只有一个线程拥有NOT_FULL是不行的
                    NOT_FULL.wait();
                }
            }
            // 保证原子性
            synchronized (LOCK_OBJECT)
            {
        
                dataList.add(element);
                amount++;
                System.out.println(Thread.currentThread().getName() + "生产了一条消息" + amount);
            }
            synchronized (NOT_EMPTY)
            {
        
                //发送未空通知
                NOT_EMPTY.notify();
            }
        }
        /**
         * 从数据区取出一个商品
         */
        public T fetch() throws Exception
        {
        
            // 数量为零，不能取元素
            while (amount <= 0)
            {
        
                synchronized (NOT_EMPTY)
                {
        
                    System.out.println(Thread.currentThread().getName() + "队列已经空了！");
                    //等待未空通知
                    NOT_EMPTY.wait();
                }
            }
            T element = null;
            // 保证原子性
            synchronized (LOCK_OBJECT)
            {
        
                element = dataList.remove(0);
                amount--;
                System.out.println(Thread.currentThread().getName() + "消费了一条消息" + amount);
            }
            synchronized (NOT_FULL)
            {
        
                //发送未满通知
                NOT_FULL.notify();
            }
            return element;
        }
    }

**生产者和消费者**

@Test
    public void testProducerConsumerQueue() throws InterruptedException {
        
        //共享数据区，实例对象
        DataBuffer<String> dataBuffer = new DataBuffer<>();
        // 同时并发执行的线程数
        final int THREAD_TOTAL = 20;
        //线程池，用于多线程模拟测试
        ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_TOTAL);
        //假定共11条线程，其中有10个消费者，但是只有1个生产者
        final int CONSUMER_TOTAL = 10;
        final int PRODUCE_TOTAL = 1;
        for (int i = 0; i < PRODUCE_TOTAL; i++) {
        
            //生产者线程每生产一个商品，间隔50毫秒
            threadPool.submit(() -> {
        
                for (int j = 0; j < 10; j ++) {
        
                    //首先生成一个随机的商品
                    String s = "商品";
                    //将商品加上共享数据区
                    try {
        
                        dataBuffer.add(s);
                    } catch (Exception e) {
        
                        throw new RuntimeException(e);
                    }
                }
            });
        }
        for (int i = 0; i < CONSUMER_TOTAL; i++)
        {
        
            //消费者线程每消费一个商品，间隔100毫秒
            threadPool.submit(() -> {
        
                for (int j = 0; j < 2; j ++) {
        
                    // 从PetStore获取商品
                    String s = null;
                    try {
        
                        s = dataBuffer.fetch();
                    } catch (Exception e) {
        
                        throw new RuntimeException(e);
                    }
                }
            });
        }
        Thread.sleep(10000);
    }

### Synchronize与wait原理 ###

`Synchronize`与`wait`都会将线程加入到等待队列中，但是两者加入的等待队列并不是同一个，`Synchronize`加入的是`对象监视器的等待队列`，当退出`Synchronize`代码块后会自动唤醒线程，而`wait`是`Object`的方法，加入的是另一个`等待集合`，只能通过`notify()`或`notifyAll()`唤醒。

## CAS和JUC原子类 ##

### CAS原理 ###

CAS(Compare And Swap)，是比较交换的缩写，可以用来实现`乐观锁`。乐观锁本质上是无锁的，每次更新前都把原来的旧值和要更新的新值一块传入，如果发现传入的旧值和当前内存上的旧值一样，则更新成功；否则更新失败；

> 乐观锁就是一直调用CAS操作，不断获取旧值，计算新值，然后传入旧值和新值进行更新，线程一直在自旋，直到更新成功为止。

**示例**

public class CompareAndSwap {
        
        public volatile int value; //值
        //不安全类
        // private static final Unsafe unsafe = Unsafe.getUnsafe();
        private static final Unsafe unsafe = getUnsafe();
        //value 的内存偏移（相对与对象头部的偏移，不是绝对偏移）
        private static final long valueOffset;
        //统计失败的次数
        public static final AtomicLong failure = new AtomicLong(0);
        static
        {
        
            try
            {
        
                //取得value属性的内存偏移
                valueOffset = unsafe.objectFieldOffset(CompareAndSwap.class.getDeclaredField("value"));
                System.out.println("valueOffset:=" + valueOffset);
            } catch (Exception ex) {
        
                throw new Error(ex);
            }
        }
        //通过CAS原子操作，进行“比较并交换”
        public final boolean unSafeCompareAndSet(int oldValue, int newValue)
        {
        
            //原子操作：使用unsafe的“比较并交换”方法进行value属性的交换
            return unsafe.compareAndSwapInt( this, valueOffset, oldValue, newValue );
        }
        //使用无锁编程实现安全的自增方法
        public void selfPlus()
        {
        
            int oldValue = value;
            //通过CAS原子操作，如果操作失败就自旋，直到操作成功
            for(;;) {
        
                oldValue = value;
                failure.incrementAndGet();
                if (unSafeCompareAndSet(oldValue, oldValue + 1)) return;
            }
            // do
            // {
        
            //     // 获取旧值
            //     oldValue = value;
            //     //统计无效的自旋次数
            //     //记录失败的次数
            //     failure.incrementAndGet();
            // } while (!unSafeCompareAndSet(oldValue, oldValue + 1));
        }
    
        /**
         * 通过反射获取Unsafe
         * @return
         */
        public static Unsafe getUnsafe()
        {
        
            try {
        
                Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
                theUnsafe.setAccessible(true);
                return (Unsafe) theUnsafe.get(null);
            } catch (Exception e) {
        
                throw new AssertionError(e);
            }
        }
    }

/**
     * 测试CAS操作
     * @throws InterruptedException
     */
    @Test
    public void testCAS() throws InterruptedException {
        
        final CompareAndSwap compareAndSwap = new CompareAndSwap();
        AtomicInteger res = new AtomicInteger(0);
        //倒数闩，需要倒数THREAD_COUNT次
        CountDownLatch latch = new CountDownLatch(10);
        for (int i = 0; i < 10; i++)
        {
        
            // 提交10个任务
                Executors.newCachedThreadPool().submit(() ->
                {
        
                    //每个任务累加1000次
                    for (int j = 0; j < 1000; j++)
                    {
        
                        compareAndSwap.selfPlus();
                        res.getAndIncrement();
                    }
                    latch.countDown();// 执行完一个任务，倒数闩减少一次
                });
        }
        latch.await();// 主线程等待倒数闩倒数完毕
        System.out.println(res);
        System.out.println("累加之和：" + compareAndSwap.value);
        System.out.println("失败次数：" + CompareAndSwap.failure.get());
    }

### `JUC`原子类 ###

`JUC`包下的原子类可以分为四组：

*  基本原子类：`AtomicInteger`，整型；`AtomicLong`，大整数；`AtomicBoolean`：布尔型；
 *  数组原子类：`AtomicIntegerArray`：整型数组原子类；`AtomicLongArray`：长整型数组原子类；`AtomicReferenceArray`：引用类型数组原子类。
 *  引用原子类：`AtomicReference`：引用类型原子类；`AtomicMarkableReference`：带有更新标记位的原子引用类型；`AtomicStampedReference`：带有更新版本号的原子引用类型。
 *  字段更新原子类：`AtomicIntegerFieldUpdater`：原子更新整型字段的更新器；`AtomicLongFieldUpdater`：原子更新长整型字段的更新器；`AtomicReferenceFieldUpdater`：原子更新引用类型里的字段。

`JUC`原子类下的底层实现也是通过不断`CAS自旋+volatile(实现可见性)`实现的，可以从源码看到。

### `ABA`问题 ###

使用`CAS自旋`更新虽然没有加锁，降低了线程切换成本，但是容易产生`ABA`问题。即线程1将值从A到B又到A，此时线程2被唤醒，以为变量没有改变过，从而引起错误的判断。解决办法是`添加时间戳`，可以借助`AtomicStampedReference`原子类实现。

## 可见性和有序性 ##

### 为什么会有可见性 ###

现代处理器都是多核的，每个核都会有自己独有的高速缓存`L1，L2，L3`，这些核又共享一个主内存，每次涉及变量更新或读取时，CPU都是先从高级缓存中读取并进行修改，然后随机写入到主存。这样就产生了问题，如果`核1`对公有变量A进行了修改，但是还没来得及写入主存，那么`核2`从主存中读取到的值就是未及时更新的脏值。

一般操作系统会使用`Lock指令`在总线上进行广播，哪些变量的高速缓存已失效，必须从主存中重新读取。Java的`volatile`关键字会在字节码上加入`loadload`、`loadstore`、`storestore`、`storeload`内存屏障来保证更改后的变量立即写入主存，且告知其他核的高速缓存该值已失效，必须从主内存重新读取。

> volatile并不保证原子性，因为虽然volatile会强制将修改刷回主存，但是修改并刷回主存的指令不是原子性的，可能有中断的可能。比如线程A修改完变量后，准备刷回主存，这时发生了线程调度，线程B知道自己的数据失效了，但是从主存中重新获取的数据不一定是最新的，因为线程A只是在本地修改了数据，但还没有写入主存。

### 参考链接 ###

[内存屏障与JVM指令][JVM]

[如果你知道这灵魂拷问的6连击，面试volitile时就稳了][6_volitile]

## 显式锁 ##

### Lock接口常用方法 ###

所有的锁实现类都会实现`Lock`接口，该接口主要有以下几个方法：

*  `lock()`：阻塞获取锁，如果当前线程不能抢到锁，线程会加入阻塞队列进行等待，直到获取到锁；
 *  `tryLock()`: 非阻塞抢锁，如果当前线程抢不到锁，线程会立刻返回`false`；
 *  `tryLock(long time, TimeUnit unit)`: 超时返回，如果当前线程在一段时间内抢不到锁，则会返回`false`；
 *  `unlock()`: 释放锁；

下面是一个使用`ReentrantLock`的示例，使用三个线程同时对某一个执行加一操作，每个线程操作100次，累计300次。

public class LockTest {
        
        private int count;
    
        @Test
        public void testReentrantLock() throws InterruptedException {
        
            Lock lock = new ReentrantLock();
    
            ExecutorService executorService = Executors.newCachedThreadPool();
    
            CountDownLatch countDownLatch = new CountDownLatch(3);
    
            for (int i = 0; i < 3; i ++) {
        
                executorService.execute(() -> {
        
                    for (int j = 0; j < 100; j ++) {
        
                        // 获取锁
                        lock.lock();
                        try {
        
                            count ++;
                        } finally {
        
                            // 释放锁
                            lock.unlock();
                        }
                    }
                    // 每完成一个线程，就更新countDownLatch
                    countDownLatch.countDown();
                });
            }
    
            countDownLatch.await();
    
            System.out.println(count); 
        }
    }

### 显式锁分类 ###

显式锁的分类有很多种，大致上可以分为下面这些：

![在这里插入图片描述][d9ab2e224f1d4a94958893b8bdba8438.png]

### 显式锁实现原理 ###

`JUC`包下的显式锁都是基于`AQS`实现的，`AQS`使用一个队列保存想要获取锁的线程，同时在队头使用`CAS`竞争获取锁，不会阻塞线程，是一种`乐观锁`。

当有新线程加入时，会通过`CAS`加入队尾，然后监控队列前一个元素的状态，这时不会发生`CAS`，但线程也不会阻塞，而是会调用`yield()`主动让出时间片。

### 参考链接 ###

[Java中常见的各种锁][Java]

[650d444dd7914fdeb8db42d035d3a4eb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/09/46bc9d5055e547b2a7ce0ad89584ddbf.png
[JVM]: https://blog.csdn.net/qq_36268025/article/details/106137960
[6_volitile]: https://zhuanlan.zhihu.com/p/359995039
[d9ab2e224f1d4a94958893b8bdba8438.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/09/3781d5ce90fd4740a4909e5faba88eb4.png
[Java]: https://blog.csdn.net/xingchensuiyue/article/details/108716466