Java处理并发编程工具集合(JUC)详解

港控/mmm° 2024-03-02 10:22 221阅读 0赞

#### 目录 ####

*  1 线程池
 *  2 Fork/Join
 *   *  2.1 概念
     *  2.2 组成
     *  2.3 基本使用
     *  2.4 设计思想
     *  2.5 注意点
 *  3 原子操作
 *   *  3.1 概念
     *  3.2 CAS
     *  3.3 atomic
     *  3.4 注意！
 *  4 AQS
 *   *  4.1 前言
     *  4.2 AQS概述
     *  4.3 基本使用
     *  4.4 原理解析
 *  5 并发容器

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

## 1 线程池 ##

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

相关文章防止冗余：

Java由浅入深理解线程池设计和原理:`https://blog.csdn.net/ZGL_cyy/article/details/133208026`  
Java线程池ExecutorService：`https://blog.csdn.net/ZGL_cyy/article/details/117843472`  
Java并发计算判断线程池中的线程是否全部执行完毕：`https://blog.csdn.net/ZGL_cyy/article/details/127599554`  
Java线程池创建方式和应用场景:`https://blog.csdn.net/ZGL_cyy/article/details/126443994`

## 2 Fork/Join ##

### 2.1 概念 ###

ForkJoinPool是由JDK1.7后提供多线程并行执行任务的框架。可以理解为一种特殊的线程池。
    
    	1.任务分割：Fork（分岔），先把大的任务分割成足够小的子任务，如果子任务比较大的话还要对子任务进行继续分割。

2.合并结果：join，分割后的子任务被多个线程执行后，再合并结果，得到最终的完整输出。

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

### 2.2 组成 ###

*  **ForkJoinTask**：主要提供fork和join两个方法用于任务拆分与合并；一般用子类 RecursiveAction（无返回值的任务）和RecursiveTask（需要返回值）来实现compute方法。
    
    ![在这里插入图片描述][fe57af6f4c5b49fc8faf633a2ab0e524.png]
 *  **ForkJoinPool**：调度ForkJoinTask的线程池；
    
    ![在这里插入图片描述][8bc9328d1dc2469e9867dd7843989785.png]
 *  **ForkJoinWorkerThread**：Thread的子类，存放于线程池中的工作线程（Worker）；
    
    ![在这里插入图片描述][93caf0d8f40749279e21e856a6b88981.png]
 *  **WorkQueue**：任务队列，用于保存任务；

### 2.3 基本使用 ###

一个典型的例子：计算1-1000的和

package com.oldlu.thread;
    
    import java.util.concurrent.*;
    
    public class SumTask {
        
    
        private static final Integer MAX = 100;
    
        static class SubTask extends RecursiveTask<Integer> {
        
            // 子任务开始计算的值
            private Integer start;
    
            // 子任务结束计算的值
            private Integer end;
    
            public SubTask(Integer start , Integer end) {
        
                this.start = start;
                this.end = end;
            }
    
            @Override
            protected Integer compute() {
        
                if(end - start < MAX) {
        
                    //小于边界，开始计算
                    System.out.println("start = " + start + ";end = " + end);
                    Integer totalValue = 0;
                    for(int index = this.start ; index <= this.end  ; index++) {
        
                        totalValue += index;
                    }
                    return totalValue;
                }else {
        
                    //否则，中间劈开继续拆分
                    SubTask subTask1 = new SubTask(start, (start + end) / 2);
                    subTask1.fork();
                    SubTask subTask2 = new SubTask((start + end) / 2 + 1 , end);
                    subTask2.fork();
                    return subTask1.join() + subTask2.join();
                }
            }
        }
    
        public static void main(String[] args) {
        
            ForkJoinPool pool = new ForkJoinPool();
            Future<Integer> taskFuture =  pool.submit(new SubTask(1,1000));
            try {
        
                Integer result = taskFuture.get();
                System.out.println("result = " + result);
            } catch (InterruptedException | ExecutionException e) {
        
                e.printStackTrace(System.out);
            }
        }
    }

### 2.4 设计思想 ###

*  普通线程池内部有两个重要集合：工作线程集合（普通线程），和任务队列。
 *  ForkJoinPool也类似，线程集合里放的是特殊线程ForkJoinWorkerThread，任务队列里放的是特殊任务ForkJoinTask
 *  不同之处在于，普通线程池只有一个队列。而ForkJoinPool的工作线程ForkJoinWorkerThread每个线程内都绑定一个双端队列。
    
    ![在这里插入图片描述][3af4ada45cc6409281b3f58e2c95635b.png]
 *  在fork的时候，也就是任务拆分，将拆分的task会被当前线程放到自己的队列中。
 *  如果有任务，那么线程优先从自己的队列里取任务执行，以`LIFO`先进后出方式从队尾获取任务，
 *  当自己队列中执行完后，工作线程会跑到其他队列以`work−stealing`窃取,窃取方式为`FIFO`先进先出，减少竞争。

### 2.5 注意点 ###

使用ForkJoin将相同的计算任务通过多线程执行。但是在使用中需要注意：

*  注意任务切分的粒度，也就是fork的界限。并非越小越好
 *  判断要不要使用ForkJoin。任务量不是太大的话，串行可能优于并行。因为多线程会涉及到上下文的切换

## 3 原子操作 ##

### 3.1 概念 ###

原子（atom）本意是“不能被进一步分割的最小粒子”，而原子操作（atomic operation）意为"不可被中断的一个或一系列操作" 。

### 3.2 CAS ###

CAS（Compare-and-Swap/Exchange），即比较并替换，是一种实现并发常用到的技术。CAS的整体架构如下：

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

juc中提供了Atomic开头的类，基于cas实现原子性操作，最基本的应用就是计数器

package com.oldlu;
    
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class AtomicCounter {
        
        private static AtomicInteger i = new AtomicInteger(0);
        public int get(){
        
            return i.get();
        }
        public void inc(){
        
            i.incrementAndGet();
        }
    
        public static void main(String[] args) throws InterruptedException {
        
            final AtomicCounter counter = new AtomicCounter();
            for (int i = 0; i < 10; i++) {
        
                new Thread(new Runnable() {
        
                    public void run() {
        
                        counter.inc();
                    }
                }).start();
            }
            Thread.sleep(3000);
            //可以正确输出10
            System.out.println(counter.i.get());
        }
    }

*注：AtomicInteger源码。基于unsafe类cas思想实现，性能篇会讲到*

CAS虽然很高效的解决了原子操作问题，但是CAS仍然存在三大问题。

1.  自旋(循环)时间长开销很大，如果CAS失败，会一直进行尝试。如果CAS长时间一直不成功，可能会给CPU带来很大的开销，注意这里的自旋是在用户态/SDK 层面实现的。
2.  只能保证一个共享变量的原子操作，对多个共享变量操作时，循环CAS就无法保证操作的原子性，这个时候就可以用锁来保证原子性。
3.  ABA问题，在使用CAS前要考虑清楚“ABA”问题是否会影响程序并发的正确性，如果需要解决ABA问题，改用传统的互斥同步可能会比CAS更高效。

### 3.3 atomic ###

上面展示了AtomicInteger，关于atomic包，还有很多其他类型：

*  基本类型
    
     *  AtomicBoolean：以原子更新的方式更新boolean；
     *  AtomicInteger：以原子更新的方式更新Integer;
     *  AtomicLong：以原子更新的方式更新Long；
 *  引用类型
    
     *  AtomicReference ： 原子更新引用类型
     *  AtomicReferenceFieldUpdater ：原子更新引用类型的字段
     *  AtomicMarkableReference ： 原子更新带有标志位的引用类型
 *  数组
    
     *  AtomicIntegerArray：原子更新整型数组里的元素。
     *  AtomicLongArray：原子更新长整型数组里的元素。
     *  AtomicReferenceArray：原子更新引用类型数组里的元素。
 *  字段
    
     *  AtomicIntegerFieldUpdater：原子更新整型的字段的更新器。
     *  AtomicLongFieldUpdater：原子更新长整型字段的更新器。
     *  AtomicStampedReference：原子更新带有版本号的引用类型。

### 3.4 注意！ ###

使用atomic要注意原子性的边界，把握不好会起不到应有的效果，原子性被破坏。

案例：原子性被破坏现象

package com.oldlu;
    
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class BadAtomic {
        
        AtomicInteger i = new AtomicInteger(0);
        static int j=0;
    
        public void badInc(){
        
            int k = i.incrementAndGet();
            try {
        
                Thread.sleep(new Random().nextInt(100));
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
            j=k;
        }
    
        public static void main(String[] args) throws InterruptedException {
        
            BadAtomic atomic = new BadAtomic();
            for (int i = 0; i < 10; i++) {
        
                new Thread(()->{
        
                    atomic.badInc();
                }).start();
            }
            Thread.sleep(3000);
            System.out.println(atomic.j);
        }
    }

结果分析：

*  每次都不一样，总之不是10
 *  i是原子性的，没问题。但是再赋值，变成了两部操作，原子性被打破
 *  在badInc上加synchronized，问题解决

## 4 AQS ##

### 4.1 前言 ###

如果要想真正的理解`JUC`下的并发工具的实现原理，我们必须要来学习`AQS`，因为它是`JUC`下很多类的基石。

在讲解AQS之前，如果老板让你自己写一个SDK层面的锁，给其他同事去使用，你会如何写呢？

1、搞一个状态标记，用来表示持有或未持有锁，但得是`volatile`类型的保证线程可见性。

2、编写一个`lock`，`unlock`函数用于抢锁和释放锁，就是对状态标记的修改操作

3、`lock`函数要保证并发下只能有一个线程能抢到锁，其他线程要等待获取锁（阻塞式），可以采用CAS+自旋的方式实现

初步实现如下：

public class MyLock {
        
        // 定义一个状态变量status：为1表示锁被持有，为0表示锁未被持有
        private volatile int status;
    
        private static final Unsafe unsafe = reflectGetUnsafe();
        private static final long valueOffset;
    
        static {
        
            try {
        
                valueOffset = unsafe.objectFieldOffset
                        (MyLock.class.getDeclaredField("status"));
            } catch (Exception ex) {
         throw new Error(ex); }
        }
    
        private static Unsafe reflectGetUnsafe() {
        
            try {
        
                Field field = Unsafe.class.getDeclaredField("theUnsafe");
                field.setAccessible(true);
                return (Unsafe) field.get(null);
            } catch (Exception e) {
        
                e.printStackTrace();
                return null;
            }
        }
    
    
        /**
         * 阻塞式获取锁
         * @return
         */
        public boolean lock() {
        
            while (!compareAndSet(0,1)) {
        
    
            }
            return true;
        }
    
        // cas 设置 status
        public final boolean compareAndSet(int expect, int update) {
        
            return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
        }
    
        /**
         * 释放锁
         */
        public void unlock() {
        
            status = 0;
        }
    }

问题：获取不到锁自旋时，是空转，浪费CPU

1、使用`yield`让出CPU执行权，等待调度

public boolean lock() {
        
        while (!compareAndSet(0,1)) {
        
            Thread.yield();//yield+自旋,尽可能的防止CPU空转,让出CPU资源
        }
        return true;
    }

或者可以采用线程休眠的方式，但是休眠时间不太好确定，太长太短都不好。

2、采用等待唤醒机制，但是这里由于没有使用`synchronized`关键字，所以也无法使用`wait/notify`，但是我们可以使用`park/unpark`，获取不到锁的线程`park`并且去队列排队，释放锁时从队列拿出一个线程`unpark`

private static final Queue<Thread> QUEUE = new LinkedBlockingQueue<>();
    
    public boolean lock() {
        
        while (!compareAndSet(0,1)) {
        
            QUEUE.offer(Thread.currentThread());
            LockSupport.park();//线程休眠
        }
        return true;
    }
    
    public void unlock() {
        
        status = 0;
        LockSupport.unpark(QUEUE.poll());
    }

### 4.2 AQS概述 ###

AQS（AbstractQueuedSynchronizer）：抽象队列同步器，定义了一套多线程访问共享资源的同步器框架，提供了SDK层面的锁机制，JUC中的很多类譬如：ReentrantLock/Semaphore/CountDownLatch…等都是基于它。

通过查阅作者的对于该类的文档注释可以得到如下核心信息：

1、AQS用一个`volatile int state;`属性表示锁状态，1表示锁被持有，0表示未被持有，具体的维护由子类去维护，但是提供了修改该属性的三个方法：`getState()`，`setState(int newState)`，`compareAndSetState(int expect, int update)`，其中CAS方法是核心。

2、框架内部维护了一个FIFO的等待队列，是用双向链表实现的，我们称之为CLH队列，

3、框架也内部也实现了条件变量`Condition`，用它来实现等待唤醒机制，并且支持多个条件变量

4、AQS支持两种资源共享的模式：独占模式（Exclusive）和共享模式（Share），所谓独占模式就是任意时刻只允许一个线程访问共享资源，譬如ReentrantLock；而共享模式指的就是允许多个线程同时访问共享资源，譬如Semaphore/CountDownLatch

5、使用者只需继承`AbstractQueuedSynchronizer`并重写指定的方法，在方法内完成对共享资源`state`的获取和释放，至于具体线程等待队列的维护，AQS已经在顶层实现好了，在那些`final`的模板方法里。

* <p>To use this class as the basis of a synchronizer, redefine the
     * following methods, as applicable, by inspecting and/or modifying
     * the synchronization state using {
        @link #getState}, {
        @link
     * #setState} and/or {
        @link #compareAndSetState}:
     *
     * <ul>
     * <li> {
        @link #tryAcquire}
     * <li> {
        @link #tryRelease}
     * <li> {
        @link #tryAcquireShared}
     * <li> {
        @link #tryReleaseShared}
     * <li> {
        @link #isHeldExclusively}
     * </ul>
     *
     * Each of these methods by default throws {
        @link
     * UnsupportedOperationException}.  Implementations of these methods
     * must be internally thread-safe, and should in general be short and
     * not block. Defining these methods is the <em>only</em> supported
     * means of using this class. All other methods are declared
     * {
        @code final} because they cannot be independently varied.

6、AQS底层使用了模板方法模式，给我们提供了许多模板方法，我们直接使用即可。

<table> 
 <thead> 
  <tr> 
   <th>API</th> 
   <th>说明</th> 
   <th></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>final void acquire(int arg)</td> 
   <td>独占模式获取锁，AQS顶层已实现，内部调用了tryAcquire</td> 
   <td>模板方法</td> 
  </tr> 
  <tr> 
   <td>boolean tryAcquire(int arg)</td> 
   <td>独占模式 尝试获取锁，AQS中未实现，由子类去实现，获取到锁返回true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>final boolean release(int arg)</td> 
   <td>释放独占锁，AQS顶层已实现，内部调用了tryRelease</td> 
   <td>模板方法</td> 
  </tr> 
  <tr> 
   <td>boolean tryRelease(int arg)</td> 
   <td>尝试释放独占锁，AQS中未实现，由子类去实现，成功释放返回true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>final void acquireShared(int arg)</td> 
   <td>共享模式获取锁，AQS顶层已实现，内部调用了tryAcquireShared</td> 
   <td>模板方法</td> 
  </tr> 
  <tr> 
   <td>int tryAcquireShared(int arg)</td> 
   <td>尝试获取共享锁，返回负数表示失败，0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源，AQS中未实现，由子类实现</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>final boolean releaseShared(int arg)</td> 
   <td>释放共享锁，返回true代表释放成功，AQS中已实现，内部调用了tryReleaseShared</td> 
   <td>模板方法</td> 
  </tr> 
  <tr> 
   <td>boolean tryReleaseShared(int arg)</td> 
   <td>尝试释放锁，释放后允许唤醒后续等待结点返回true，否则返回false，AQS中未实现，需要由子类实现</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>boolean isHeldExclusively()</td> 
   <td>共享资源是否被独占</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

### 4.3 基本使用 ###

此时老板给你加了需求，要求你实现一个基于AQS的锁，那该怎么办呢？

在`AbstractQueuedSynchronizer`的类注释中给出了使用它的基本方法，我们按照它的写法尝试即可

/**
     * 基于 aqs实现锁
     */
    public class MyLock implements Lock {
        
    
        //同步器
        private Syn syn = new Syn();
    
        @Override
        public void lock() {
        
            //调用模板方法
            syn.acquire(1);
        }
    
        @Override
        public void unlock() {
        
            //调用模板方法
            syn.release(0);
        }
    
        // 其他接口方法暂时先不实现 省略
    
    
        // 实现一个独占同步器
        class Syn extends AbstractQueuedSynchronizer{
        
            @Override
            protected boolean tryAcquire(int arg) {
        
                if (compareAndSetState(0,arg)) {
        
                    return true;
                }
                return false;
            }
    
            @Override
            protected boolean tryRelease(int arg) {
        
                setState(arg);
                return true;
            }
    
        }
    
    }

### 4.4 原理解析 ###

自己实现的锁在使用过程中发现一个问题，就是有时候有的线程特别容易抢到锁，而有的线程老是抢不到锁，虽说线程们抢锁确实看命，但能不能加入一种设计，让各个线程机会均等些，起码不要出现某几个线程总是特倒霉抢不到锁的情况吧！

这其实就是涉及到锁是否是公平的，那么什么是公平锁什么是非公平锁呢？

这我们就不得不深入我们使用的模板方法中看一眼了

/**
         * Acquires in exclusive mode, ignoring interrupts.  Implemented
         * by invoking at least once {@link #tryAcquire},
         * returning on success.  Otherwise the thread is queued, possibly
         * repeatedly blocking and unblocking, invoking {@link
         * #tryAcquire} until success.  This method can be used
         * to implement method {@link Lock#lock}.
         *
         * @param arg the acquire argument.  This value is conveyed to
         *        {@link #tryAcquire} but is otherwise uninterpreted and
         *        can represent anything you like.
         */
    public final void acquire(int arg) {
        
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    
    //结合我自己写的尝试获取锁的方法
    protected boolean tryAcquire(int arg) {
        
        if (compareAndSetState(0,arg)) {
        
            return true;
        }
        return false;
    }

这里大概描述如下：

1、线程一来首先调用`tryAcquire`，在`tryAcquire`中直接CAS获取锁，如果获取不成功通过`addWaiter`加入等待队列，然后走`acquireQueued`让队列中的某个等待线程去获取锁。

2、不公平就体现在这里，线程来了也不先看一下等待队列中是否有线程在等待，如果没有线程等待，那直接获取锁没什么 问题，如果有线程等待就直接去获取锁不就相当于插队么？

那如何实现这种公平性呢？这就不得不探究一下AQS的内部的实现原理了，下面我们依次来看：

1、查看`AbstractQueuedSynchronizer`的类定义，虽然它里面代码很多，但重要的属性就那么几个，

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
        implements java.io.Serializable {
        
        
        private volatile int state;
        private transient volatile Node head;
        private transient volatile Node tail;
        
        static final class Node {
        
            //其他不重要的略
            volatile int waitStatus;
            volatile Node prev;
            volatile Node next;
            volatile Thread thread;
        }
        
        public class ConditionObject implements Condition, java.io.Serializable {
        ...}
    }

结合前面讲的AQS的类文档注释不难猜到，内部类 `Node`以及其类型的变量 `head` 和 `tail` 就表示 AQS 内部的一个等待队列，而剩下的 `state` 变量就用来表示锁的状态。

等待队列应该就是线程获取锁失败时，需要临时存放的一个地方，用来等待被唤醒并尝试获取锁。再看 `Node` 的属性我们知道，`Node` 存放了当前线程的指针 `thread`，也即可以表示当前线程并对其进行某些操作，`prev` 和 `next` 说明它构成了一个双向链表，也就是为某些需要得到前驱或后继节点的算法提供便利。

2、AQS加锁最核心的代码就是如下，我们要来探究它的实现原理

public final void acquire(int arg) {
        
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

它的原理及整个过程我们以图的形式说明如下：

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

3、原理搞懂了，那如何让自定义的锁是公平的呢？

其实导致不公平的原因就是线程每次调用`acquire`时，都会先去`tryAcquire`，而该方法目前的实现时直接去抢锁，也不看现在等待队列中有没有线程在排队，如果有线程在排队，那岂不是变成了插队，导致不公平。

所以现在的解决办法就是，在`tryAcquire`时先看一下等待队列中是否有在排队的，如果有那就乖乖去排队，不插队，如果没有则可以直接去获取锁。

那如何知道线程AQS等待队列中是否有线程排队呢？其实AQS顶层已经实现好了，它提供了一个`hasQueuedPredecessors`函数：如果在当前线程之前有一个排队的线程，则为True; 如果当前线程位于队列的头部(`head.next`)或队列为空，则为false。

protected boolean tryAcquire(int arg) {
        
        //先判断等待队列中是否有线程在排队 没有线程排队则直接去获取锁
        if (!hasQueuedPredecessors() && compareAndSetState(0,arg)) {
        
            return true;
        }
        return false;
    }

4、现在已经有公平锁了，但是成年人的世界不是做选择题，而是都想要，自己编写的锁既能支持公平锁，也支持非公平锁，让使用者可以自由选择，怎么办？

其实只要稍微改造一下即可，

5、现在锁的公平性问题解决了，但是老板又出了新的需求，要求我们的锁支持可重入，因为它写了如下一段代码，发现一直获取不到锁

static Lock lock = new MyLock();
    static void test3() {
        
        lock.lock();
        try {
        
            System.out.println("test3 get lock,then do something ");
            test4();
        } finally {
        
            lock.unlock();
        }
    }
    static void test4() {
        
        lock.lock();
        try {
        
            System.out.println("test4 get lock,then do something ");
        } finally {
        
            lock.unlock();
        }
    }

那如何让锁支持可重入呢？也就是说如果一个线程持有锁之后，还能继续获取锁，也就是说让锁只对不同线程互斥。

查看`AbstractQueuedSynchronizer`的定义我们发现，它还继承自另一个类：`AbstractOwnableSynchronizer`

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
        implements java.io.Serializable {
        ...}
    
    public abstract class AbstractOwnableSynchronizer
        implements java.io.Serializable {
        
        
        private transient Thread exclusiveOwnerThread;
        
        protected final void setExclusiveOwnerThread(Thread thread) {
        ...}
        protected final Thread getExclusiveOwnerThread(){
        ...}
    }

看到这我们明白了，原来`AQS`中有个变量是可以保存当前持有独占锁的线程的。那好办了，当我们获取锁时，如果发现锁被持有不要着急放弃，先看看持有锁的线程是否时当前线程，如果是还能继续获取锁。

另外关于可重入锁，还要注意一点，锁的获取和释放操作是成对出现的，就像下面这样

lock
        lock
            lock
                lock
                    ....
                unlock
            unlock
        unlock
    unlock

所以对于重入锁不仅要能记录锁被持有，还要记录重入的次数，释放的时候也不是直接将锁真实的释放，而是先减少重入次数，能释放的时候在释放。

故此时状态变量`state`不在只有两个取值`0,1`，某线程获取到锁`state=1`，如果当前线程重入获取只需增加状态值`state=2`，依次同理，锁释放时释放一次状态值`-1`，当`state=0`时才真正释放，其他线程才能继续获取锁。

修改我们锁的代码如下：公平非公平在可重入上的逻辑是一样的

public class MyLock implements Lock {
        
    
        //同步器
        private Sync syn ;
    
        MyLock () {
        
            syn = new NoFairSync();
        }
    
        MyLock (boolean fair) {
        
            syn = fair ? new FairSync():new NoFairSync();
        }
    
    
        @Override
        public void lock() {
        
            //调用模板方法
            syn.acquire(1);
        }
    
        @Override
        public void unlock() {
        
            //调用模板方法
            syn.release(1);
        }
    
        // Lock接口其他方法暂时先不实现 略
    
    
        // 实现一个独占同步器
        class Sync extends AbstractQueuedSynchronizer{
        
    
    
            @Override
            protected boolean tryRelease(int arg) {
        
                if (Thread.currentThread() != getExclusiveOwnerThread()) {
        
                    throw new IllegalMonitorStateException();
                }
                boolean realRelease = false;
                int nextState = getState() - arg;
                if (nextState == 0) {
        
                    realRelease = true;
                    setExclusiveOwnerThread(null);
                }
                setState(nextState);
                return realRelease;
            }
    
        }
    
        class FairSync extends Sync {
        
            @Override
            protected boolean tryAcquire(int arg) {
        
                final Thread currentThread = Thread.currentThread();
                int currentState = getState();
                if (currentState == 0 ) {
         // 可以获取锁
                    //先判断等待队列中是否有线程在排队 没有线程排队则直接去获取锁
                    if (!hasQueuedPredecessors() && compareAndSetState(0,arg)) {
        
                        setExclusiveOwnerThread(currentThread);
                        return true;
                    }
                }else if (currentThread == getExclusiveOwnerThread()) {
        
                    //重入逻辑 增加 state值
                    int nextState = currentState + arg;
                    if (nextState < 0) {
        
                        throw new Error("Maximum lock count exceeded");
                    }
                    setState(nextState);
                    return true;
                }
                return false;
            }
        }
    
        class NoFairSync extends Sync {
        
            @Override
            protected boolean tryAcquire(int arg) {
        
                final Thread currentThread = Thread.currentThread();
                int currentState = getState();
                if (currentState ==0 ) {
         // 可以获取锁
                    //直接去获取锁
                    if (compareAndSetState(0,arg)) {
        
                        setExclusiveOwnerThread(currentThread);
                        return true;
                    }
                }else if (currentThread == getExclusiveOwnerThread()) {
        
                    //重入逻辑 增加 state值
                    int nextState = currentState + arg;
                    if (nextState < 0) {
        
                        throw new Error("Maximum lock count exceeded");
                    }
                    setState(nextState);
                    return true;
                }
                return false;
            }
        }
    
    }

好了至此我们已经掌握了`AQS`的核心原理以及它的一个经典实现`ReentrantLock`几乎全部的知识点，此时打开`ReentrantLock`的源码你会发现一切都很清爽！！！

## 5 并发容器 ##

juc中还包含很多其他的并发容器（了解）

**1.ConcurrentHashMap**

对应：HashMap

目标：代替Hashtable、synchronizedMap，使用最多，源码篇会详细讲解

原理：JDK7中采用Segment分段锁，JDK8中采用CAS+synchronized

**2.CopyOnWriteArrayList**

对应：ArrayList

目标：代替Vector、synchronizedList

原理：高并发往往是读多写少的特性，读操作不加锁，而对写操作加Lock独享锁，先复制一份新的集合，在新的集合上面修改，然后将新集合赋值给旧的引用，并通过volatile 保证其可见性。

*查看源码：volatile array，lock加锁，数组复制*

**3.CopyOnWriteArraySet**

对应：HashSet

目标：代替synchronizedSet

原理：与CopyOnWriteArrayList实现原理类似。

**4.ConcurrentSkipListMap**

对应：TreeMap

目标：代替synchronizedSortedMap(TreeMap)

原理：基于Skip list（跳表）来代替平衡树，按照分层key上下链接指针来实现。

*附加：跳表*

**5.ConcurrentSkipListSet**

对应：TreeSet

目标：代替synchronizedSortedSet(TreeSet)

原理：内部基于ConcurrentSkipListMap实现，原理一致

**6.ConcurrentLinkedQueue**

对应：LinkedList

对应：无界线程安全队列

原理：通过队首队尾指针，以及Node类元素的next实现FIFO队列

**7.BlockingQueue**

对应：Queue

特点：拓展了Queue，增加了可阻塞的插入和获取等操作

原理：通过ReentrantLock实现线程安全，通过Condition实现阻塞和唤醒

实现类：

*  LinkedBlockingQueue：基于链表实现的可阻塞的FIFO队列
 *  ArrayBlockingQueue：基于数组实现的可阻塞的FIFO队列
 *  PriorityBlockingQueue：按优先级排序的队列

[c8c8087edff54fdab5f01c93594ca89f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/41e627079b8b4158aab33e2a78edd902.png
[6ddfa50e6699493db8357471bca38ff0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/997cc9be992a453db25c414f6625bbd4.png
[fe57af6f4c5b49fc8faf633a2ab0e524.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/2675e0f071c64a44b474707b43203b74.png
[8bc9328d1dc2469e9867dd7843989785.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/e5b54e15aefe40b6af0247a89e1a57a8.png
[93caf0d8f40749279e21e856a6b88981.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/da9a446c1a21408ba143659721736410.png
[3af4ada45cc6409281b3f58e2c95635b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/943524f2ec7e46ce87170cd6871e2142.png
[6568e04dd21847059f67fe06f24fa4da.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/fb2267deff984c91878a2794c868f468.png
[fa8a5a3b80e44e47aec138ab46c1618f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/afd5979cb7b8485eb3d674b31d32c90f.png