认识达内从这里开始

认真做教育专心促就业

Java 编程之美：并发编程高级篇（上）

发布：达内科技
来源：互联网
时间：2018-07-03 14:48

    一、前言-达内科技负责整理，你们要仔细看喔
    借用 Java 并发编程实践中的话:编写正确的程序并不容易,而编写正常的并发程序就更难了.
    相比于顺序执行的情况,多线程的线程安全问题是微妙而且出乎意料的,因为在没有进行适当同步的情况下多线程中各个操作的顺序是不可预期的.
    并发编程相比 Java 中其他知识点学习起来门槛相对较高,学习起来比较费劲,从而导致很多人望而却步;
    而无论是职场面试和高并发高流量的系统的实现却都还离不开并发编程,从而导致能够真正掌握并发编程的人才成为市场比较迫切需求的.
    本 Chat 作为 Java 并发编程之美系列的高级篇之三,主要讲解锁,内容如下:(建议先阅读 Java 并发编程之美:并发编程高级篇之二 )
    抽象同步队列 AQS (AbstractQueuedSynchronizer) 概述,AQS 是实现同步的基础组件,并发包中锁的实现底层就是使用 AQS 实现,虽然大多数开发者可能从来不会直接用到 AQS,但是知道其原理对于架构设计还是很有帮助的.
    独占锁 ReentrantLock 原理探究,ReentrantLock 是可重入的独占锁或者叫做排它锁,同时只能有一个线程可以获取该锁,其实现分为公平与非公平的独占锁.
    读写锁 ReentrantReadWriteLock 原理,ReentrantLock 是独占锁,同时只有一个线程可以获取该锁,而实际情况下会有写少读多的场景,显然 ReentrantLock 满足不了需求,所以 ReentrantReadWriteLock 应运而生,本文来介绍读写分离锁的实现.
    StampedLock 锁原理探究,StampedLock 是并发包里面 jdk8 版本新增的一个锁,该锁提供了三种模式的读写控制.
    二、抽象同步队列 AQS 概述
    2.1 AQS - 锁的底层支持
    AbstractQueuedSynchronizer 抽象同步队列, 简称 AQS,是实现同步器的基础组件,并发包中锁的实现底层就是使用 AQS 实现,另外大多数开发者可能从来不会直接用到 AQS,但是知道其原理对于架构设计还是很有帮助的
    AQS 是一个 FIFO 的双向队列,内部通过节点 head 和 tail 记录队首和队尾元素,队列元素类型为 Node.
    其中 Node 中 thread 变量用来存放进入 AQS 队列里面的线程;Node 节点内部 SHARED 用来标记该线程是获取共享资源时候被阻塞挂起后放入 AQS 队列,EXCLUSIVE 标示线程是获取独占资源时候被挂起后放入 AQS 队列;
    waitStatus 记录当前线程等待状态,分别为 CANCELLED(线程被取消了),SIGNAL(线程需要被唤醒),CONDITION(线程在条件队列里面等待),PROPAGATE(释放共享资源时候需要通知其它节点);prev 记录当前节点的前驱节点,next 记录当前节点后继节点.
    AQS 中维持了一个单一的状态信息 state, 可以通过 getState,setState,compareAndSetState 函数修改其值;对于 ReentrantLock 的实现来说,state 可以用来表示当前线程获取锁的可重入次数;
    对应读写锁 ReentrantReadWriteLock 来说 state 的高 16 位表示读状态也就是获取该读锁的次数,低 16 位表示获取到写锁的线程的可重入次数;对于 semaphore 来说 state 用来表示当前可用信号的个数;
    对于 FutuerTask 来说,state 用来表示任务状态(例如还没开始,运行,完成,取消);对应 CountDownlatch 和 CyclicBarrie 来说 state 用来表示计数器当前的值.
    达内科技的老师讲解这里：AQS 有个内部类 ConditionObject 是用来结合锁实现线程同步,ConditionObject 可以直接访问 AQS 对象内部的变量,比如 state 状态值和 AQS 队列;
    ConditionObject 是条件变量,每个条件变量对应着一个条件队列 (单向链表队列),用来存放调用条件变量的 await() 方法后被阻塞的线程,如类图,这个条件队列的头尾元素分别为 firstWaiter 和 lastWaiter.
    对于 AQS 来说线程同步的关键是对状态值 state 进行操作,根据 state 是否属于一个线程,操作 state 的方式分为独占模式和共享模式.
    独占方式下获取和释放资源使用方法为:
    void acquire(int arg)void acquireInterruptibly(int arg)boolean release(int arg)
    共享模式下获取和释放资源方法为:
    void acquireShared(int arg)void acquireSharedInterruptibly(int arg)boolean releaseShared(int arg)
    对于独占方式获取的资源是与具体线程绑定的,就是说如果一个线程获取到了资源,就会标记是那个线程获取到了,其它线程尝试操作 state 获取资源时候发现当前该资源不是自己持有的,就会获取失败后被阻塞;
    比如独占锁 ReentrantLock 的实现,当一个线程获取了 ReentrantLock 的锁后,AQS 内部会首先使用 CAS 操作把 state 状态值从 0 变为 1,然后设置当前锁的持有者为当前线程.
    当该线程再次获取锁时候发现当前线程就是锁的持有者则会把状态值从 1 变为 2 也就是设置可重入次数,当另外一个线程获取锁的时候发现自己并不是该锁的持有者就会被放入 AQS 阻塞队列后挂起.
    对应共享操作方式资源是与具体线程不相关的,多个线程去请求资源时候是通过 CAS 方式竞争获取资源,当一个线程获取到了资源后,另外一个线程再次获取时候如果当前资源还能满足它的需要,则当前线程只需要使用 CAS 方式进行获取即可,共享模式下并不需要记录那个线程获取了资源;
    比如 Semaphore 信号量,当一个线程通过 acquire() 方法获取一个信号量时候,会首先看当前信号量个数是否满足需要,不满足则把当前线程放入阻塞队列,如果满足则通过自旋 CAS 获取信号量.
    对应独占模式的获取与释放资源流程:
    1)当一个线程调用 acquire(int arg) 方法获取独占资源时候,会首先使用 tryAcquire 尝试获取资源,具体是设置状态变量 state 的值,成功则直接返回.失败则将当前线程封装为类型为 Node.EXCLUSIVE 的 Node 节点后插入到 AQS 阻塞队列尾部,并调用 LockSupport.park(this) 挂起当前线程.
    public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }

2)当一个线程调用 release(int arg) 时候会尝试使用, tryRelease 操作释放资源,这里是设置状态变量 state 的值,然后调用 LockSupport.unpark(thread) 激活 AQS 队列里面最早被阻塞的线程 (thread).被激活的线程则使用 tryAcquire 尝试看当前状态变量 state 的值是否能满足自己的需要,满足则该线程被激活然后继续向下运行,否者还是会被放入 AQS 队列并被挂起.

达内科技

    达内科技的老师说，这块得仔细看！public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; }
    需要注意的 AQS 类并没有提供可用的 tryAcquire 和 tryRelease,正如 AQS 是锁阻塞和同步器的基础框架,tryAcquire 和 tryRelease 需要有具体的子类来实现.
    子类在实现 tryAcquire 和 tryRelease 时候要根据具体场景使用 CAS 算法尝试修改状态值 state, 成功则返回 true, 否者返回 false.
    子类还需要定义在调用 acquire 和 release 方法时候 state 状态值的增减代表什么含义.
    比如继承自 AQS 实现的独占锁 ReentrantLock,定义当 status 为 0 的时候标示锁空闲,为 1 的时候标示锁已经被占用,在重写 tryAcquire 时候,内部需要使用 CAS 算法看当前 status 是否为 0,如果为 0 则使用 CAS 设置为 1,并设置当前线程的持有者为当前线程,并返回 true, 如果 CAS 失败则返回 false.
    比如继承自 AQS 实现的独占锁实现 tryRelease 时候,内部需要使用 CAS 算法把当前 status 值从 1 修改为 0,并设置当前锁的持有者为 null,然后返回 true, 如果 cas 失败则返回 false.
    对应共享资模式的获取与释放流程:
    1)当线程调用 acquireShared(int arg) 获取共享资源时候,会首先使用 tryAcquireShared 尝试获取资源,具体是设置状态变量 state 的值,成功则直接返回.失败则将当前线程封装为类型为 Node.SHARED 的 Node 节点后插入到 AQS 阻塞队列尾部,并使用 LockSupport.park(this) 挂起当前线程.
    public final void acquireShared(int arg) { if (tryAcquireShared(arg) < 0) doAcquireShared(arg); }
    2)当一个线程调用 releaseShared(int arg) 时候会尝试使用, tryReleaseShared 操作释放资源,这里是设置状态变量 state 的值,然后使用 LockSupport.unpark(thread)激活 AQS 队列里面最早被阻塞的线程 (thread).
    被激活的线程则使用 tryReleaseShared 尝试看当前状态变量 state 的值是否能满足自己的需要,满足则该线程被激活然后继续向下运行,否者还是会被放入 AQS 队列并被挂起.
    public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }
    同理需要注意的 AQS 类并没有提供可用的 tryAcquireShared 和 tryReleaseShared,正如 AQS 是锁阻塞和同步器的基础框架,tryAcquireShared 和 tryReleaseShared 需要有具体的子类来实现.
    子类在实现 tryAcquireShared 和 tryReleaseShared 时候要根据具体场景使用 CAS 算法尝试修改状态值 state, 成功则返回 true, 否者返回 false.
    比如继承自 AQS 实现的读写锁 ReentrantReadWriteLock 里面的读锁在重写 tryAcquireShared 时候,首先看写锁是否被其它线程持有,如果是则直接返回 false,否者使用 cas 递增 status 的高 16 位,在 ReentrantReadWriteLock 中 status 的高 16 为获取读锁的次数.
    比如继承自 AQS 实现的读写锁 ReentrantReadWriteLock 里面的读锁在重写 tryReleaseShared 时候,内部需要使用 CAS 算法把当前 status 值的高 16 位减一,然后返回 true, 如果 cas 失败则返回 false.
    基于 AQS 实现的锁除了需要重写上面介绍的方法,还需要重写 isHeldExclusively 方法用来判断锁是被当前线程独占还是被共享.
    另外也许你会好奇独占模式下的
    void acquire(int arg)void acquireInterruptibly(int arg)
    和共享模式下获取资源的:
    void acquireShared(int arg) void acquireSharedInterruptibly(int arg)
    这两套函数都有一个带有 Interruptibly 关键字的函数,那么带有这个关键字的和不带的有什么区别那?
    其实不带 Interruptibly 关键字的方法是说不对中断进行响应,也就是线程在调用不带 Interruptibly 关键字的方法在获取资源的时候或者获取资源失败被挂起时候,其他线程中断了该线程,那么该线程不会因为被中断而抛出异常,还是继续获取资源或者被挂起,也就是不对中断进行响应,忽略中断.
    带 Interruptibly 关键字的方法是说对中断进行响应,也就是也就是线程在调用带 Interruptibly 关键字的方法在获取资源的时候或者获取资源失败被挂起时候,其他线程中断了该线程,那么该线程会抛出 InterruptedException 异常而返回.
    本节最后我们来看看 AQS 提供的队列是如何维护的,主要看入队操作
    入队操作: 当一个线程获取锁失败后该线程会被转换为 Node 节点,然后就会使用 enq(final Node node) 方法插入该节点到 AQS 的阻塞队列,
    private Node enq(final Node node) { for (;;) { Node t = tail;//(1) if (t == null) { // Must initialize if (compareAndSetHead(new Node()))//(2) tail = head; } else { node.prev = t;//(3) if (compareAndSetTail(t, node)) {//(4) t.next = node; return t; } } } }
    下面结合代码和下面的节点图来讲解下,如上代码第一次循环当要在 AQS 队列尾部插入元素时候,AQS 队列状态为图(default), 也就是队列头尾节点都指向 null;当执行代码(1)后节点 t 指向了尾部节点,这时候队列状态如图(I).
    可知这时候 t 为 null,则执行代码(2)使用 CAS 算法设置一个哨兵节点为头结点,如果 CAS 设置成功,然后让尾部节点也指向哨兵节点,这时候队列状态如图(II).
    到现在只是插入了一个哨兵节点,还需要插入的 node 节点,所以第二次循环后执行到步骤(1),这时候队列状态如图(III);然后执行代码(3)设置 node 的前驱节点为尾部节点,这时候队列状态图如图(IV);然后通过 CAS 算法设置 node 节点为尾部节点,CAS 成功后队列状态图为(V);CAS 成功后在设置原来的尾部节点的后驱节点为 node, 这时候就完成了双向链表的插入了,这时候队列状态为图(VI).
    2.2 AQS - 条件变量的支持
    正如基础篇讲解的 notify 和 wait 是配合 synchronized 内置锁实现线程间同步基础设施,条件变量的 signal 和 await 方法是用来配合锁(使用 AQS 实现的锁)实现线程间同步的基础设施.
    在基础篇讲解了在调用共享变量的 notify 和 wait 方法前必须先获取该共享变量的内置锁,同理在调用条件变量的 signal 和 await 方法前必须先获取条件变量对应的锁.
    说了那么多,到底什么是条件变量那?如何使用那?不急,下面看一个例子:
    ReentrantLock lock = new ReentrantLock();//(1)Condition condition = lock.newCondition();//(2)lock.lock();//(3)try { System.out.println("begin wait"); condition.await();//(4) System.out.println("end wait");} catch (Exception e) { e.printStackTrace();} finally { lock.unlock();//(5)}lock.lock();//(6)try { System.out.println("begin signal"); condition.signal();//(7) System.out.println("end signal");} catch (Exception e) { e.printStackTrace();} finally { lock.unlock();//(8)}
    如上代码(1)创建了一个独占锁 ReentrantLock 的对象,ReentrantLock 是基于 AQS 实现的锁.
    代码(2)使用创建的 lock 对象的 newCondition()方法创建了一个 ConditionObject 变量,这个变量就是 lock 锁对应的一个条件变量.需要注意的是一个 Lock 对象可以创建多个条件变量.
    以上就是达内科技给大家做的内容详解,更多关于IT知识的学习,请继续关注达内科技

< 上一篇：Java 编程之美：并发编程基础（下）

下一篇：Java 编程之美：并发编程高级篇（中） >