`

AQS

 
阅读更多

谢谢原作者 ： singleAnt,写的非常好【java并发】juc高级锁机制探讨

 

1. Java中 的高级锁 ：

   RentrantLock  :// 可重入的锁 ，它是一种独占锁 ， 只能有一个对象拥有该锁，区别于synchronize，可中断 ，可定时，可公平的获得锁

   RentrantReadWriteLock ： // 可重入的读写锁，多个线程可以统一同时读，读是共享的，但写是独占的

   Semaphore ： // 信号量，它是一个共享锁，可以把他想象成一个池，里面放着当前可以使用的资源，多所有线程都是一样的，没有谁独占

   CountDownLatch ： // 共享锁 ，只有当count -> 0 , 那么 await()才允许线程通过

   CyclicBarrier ： // 共享锁 ， 与CountDownLatch 类似，它是累加至相应的paties , 才通过

 

2. 这些锁的共性 : // 模板方法

   多线程并发的执行，之间通过某种 共享 状态来同步，只有当状态满足 xxxx 条件，才能触发线程执行 xxxx ；

   像ReentrantLock ,共享状态就是对象的锁的是否被某个线程占用着，如果没有，那么当前线程占用，把state = 1

   如果占用着，那么就看是不是这个线程占用着的，是的话，就将state + 1 , 如果不是就加入工作队列 ，wait,

   在就如，CountDownLatech , 共享状态，就是count值，每一次调用countDown() ,count-- ， 然后就会检查是否==0，

   如果== 0 ，就会唤醒所有线程！

  所以，把这个中间的过程抽象出来，定义骨架，然后具体实现，就各个锁（子类）自己去实现 === 这不就是模板方法吗？

  而这个骨架 就是AQS框架，这些锁都是基于他创建的

 下面来看ReentrantLock的基本原理：

ReentrantLock原理

由于同步器里已经定义了基本的结构，包括获取、释放、和阻塞队列维护和管理等。ReentrantLock是一个独占互斥锁，里只需要实现TryAcquire、TryRelease等方法，告诉同步器是否获取和释放状态成功。其他的后续行为都由AQS框架完成。由于ReentrantLock是一个可重入的独占锁，所以同步器状态可以直接根据是否==0来判断是否可用。

ReentrantLock主要提供lock和unlcok两个方法。

而lock和unlock正是基于AQS的一个子类同步器来实现。里面sync同步器有两种实现，一种是公平锁，一种是非公平锁。默认是非公平锁，看看非公平锁实现tryAcquire

Java代码   收藏代码
  1. final boolean nonfairTryAcquire(int acquires) {  
  2.  final Thread current = Thread.currentThread();  
  3.  int c = getState();  
  4.  if (c == 0) {//如果状态位为0，那么尝试获取  
  5.      if (compareAndSetState(0, acquires)) {//基于CAS获取和修改状态  
  6.          setExclusiveOwnerThread(current);//成功则设置当前线程为独占执行线程  
  7.          return true;  
  8.      }  
  9.  }  
  10.  else if (current == getExclusiveOwnerThread()) {//当前线程已是执行线程  
  11.      int nextc = c + acquires;//累加  
  12.      if (nextc < 0) // overflow  
  13.          throw new Error("Maximum lock count exceeded");  
  14.      setState(nextc);  
  15.      return true;  
  16.  }  
  17.  return false;//其他情况下代表获取失败  

 

再看看公平锁的tryAcquire

 

Java代码   收藏代码
  1.     protected final boolean tryAcquire(int acquires) {  
  2.     final Thread current = Thread.currentThread();  
  3.     int c = getState();  
  4.     if (c == 0) {  
  5.         if (isFirst(current) &&  
  6.             compareAndSetState(0, acquires)) {//判断是否是第一个线程，是的话才尝试获取锁  
  7.             setExclusiveOwnerThread(current);  
  8.             return true;  
  9.         }  
  10.     }  
  11.     else if (current == getExclusiveOwnerThread()) {  
  12.         int nextc = c + acquires;  
  13.         if (nextc < 0)  
  14.             throw new Error("Maximum lock count exceeded");  
  15.         setState(nextc);  
  16.         return true;  
  17.     }  
  18.     return false;  
  19. }  

 可以看看 非公平锁的不会根据FIFO，而公平锁会判断是否是第一个线程，根据FIFO来执行。

中间的调用过程 ：

  lock.lock()调用 NofairSync.lock() ，他就会检查，锁是否被占用了compareAndSetState(0,1),如果没有则

  当前线程占有锁，如果占用了，调用AQS的acquire()方法，看获得锁的线程是不是当前线程，调用NoFairSync的tryAcquire(),是当前线程，重新设置state，返回true,否则，则会acquireQueue(),多的当前阻塞队列，入队！可能你会觉得当我们向他加锁的过程，要经过这么多步骤，而中间没有synchronize同步，那么，别的线程也可以并发的访问，怎么能做到加锁了？关键就在于volatile state变量 和  compareAndSetState() /CAS

他们保证了线程间的通信，也就是说，那个线程访问到了CAS这一步，他对State的操作，立马就能够对其他线程看到！保证这有一个线程得到锁！

 

 与ReentrantLock有关的类：

  

 

 

3. 共享锁 和 独占锁

   独占锁模式下，每次只能有一个线程能持有锁，ReentrantLock就是以独占方式实现的互斥锁。共享锁，则允许多个线程同时获取锁，并发访问 共享资源，如：ReadWriteLock。AQS的内部类Node定义了两个常量SHARED和EXCLUSIVE，他们分别标识 AQS队列中等待线程的锁获取模式。

 

4. 锁的公平性：

   锁的公平与非公平，是指线程请求获取锁的过程中，是否允许插队。在公平锁上，线程将按他们发出请求的顺序来获得锁；而非公平锁则允许在线程发出请求后立即尝试获取锁，如果可用则可直接获取锁，尝试失败才进行排队等待。ReentrantLock提供了两种锁获取方式，FairSyn和NofairSync。结论：ReentrantLock是以独占锁的加锁策略实现的互斥锁，同时它提供了公平和非公平两种锁获取方式。

 

5.AQS框架:

  

AbstractQueuedSynchronizer 是一个抽象类，里面定义了同步器的基本框架，实现了基本的结构功能。只留有状态条件的维护由具体同步器根据具体场景来定制，如上面提到的 ReentrantLock 、 RetrantReadWriteLock和CountDownLatch 等等。

一个同步器至少需要包含两个功能：

1.       获取同步状态

如果允许，则获取锁，如果不允许就阻塞线程，直到同步状态允许获取。

2.       释放同步状态

修改同步状态，并且唤醒等待线程。

根据作者论文， aqs 同步机制同时考虑了如下需求：

1.       独占锁和共享锁两种机制。

2.       线程阻塞后，如果需要取消，需要支持中断。

3.       线程阻塞后，如果有超时要求，应该支持超时后中断的机制。

 

实现涉及基本技术原理  

1.       状态位   

提供 volatile 变量 state;  用于同步线程之间的共享状态。通过 CAS 和 volatile 保证其原子性和可见性。对应源码里的定义：

Java代码   收藏代码
  1. /** 
  2.  * 同步状态 
  3.  */  
  4. private volatile int state;  
  5.   
  6. /** 
  7.  *cas 
  8.  */  
  9. protected final boolean compareAndSetState(int expect, int update) {  
  10.     // See below for intrinsics setup to support this  
  11.     return unsafe.compareAndSwapInt(this, stateOffset, expect, update);  
  12. }  

 

2.       线程阻塞和唤醒

有别于wait和notiry。这里利用 jdk1.5 开始提供的 LockSupport.park() 和 LockSupport.unpark() 的本地方法实现，实现线程的阻塞和唤醒。

3.       阻塞线程节点队列 CHL Node queue 。

根据论文里描述， AQS 里将阻塞线程封装到一个内部类 Node 里。并维护一个 CHL Node FIFO 队列。 CHL队列是一个非阻塞的 FIFO 队列，也就是说往里面插入或移除一个节点的时候，在并发条件下不会阻塞，而是通过自旋锁和 CAS 保证节点插入和移除的原子性。实现无锁且快速的插入。关于非阻塞算法可以参考  Java 理论与实践: 非阻塞算法简介 。CHL队列对应代码如下：

 

Java代码   收藏代码
  1.  /** 
  2.  * CHL头节点 
  3.  */   
  4. rivate transient volatile Node head;  
  5. /** 
  6.  * CHL尾节点 
  7.  */  
  8. private transient volatile Node tail;  

  Node节点是对Thread的一个封装，结构大概如下：

Java代码   收藏代码
  1. static final class Node {  
  2.     /** 代表线程已经被取消*/  
  3.     static final int CANCELLED =  1;  
  4.     /** 代表后续节点需要唤醒 */  
  5.     static final int SIGNAL    = -1;  
  6.     /** 代表线程在等待某一条件/ 
  7.     static final int CONDITION = -2; 
  8.     /** 标记是共享模式*/  
  9.     static final Node SHARED = new Node();  
  10.     /** 标记是独占模式*/  
  11.     static final Node EXCLUSIVE = null;  
  12.   
  13.     /** 
  14.      * 状态位 ，分别可以使CANCELLED、SINGNAL、CONDITION、0 
  15.      */  
  16.     volatile int waitStatus;  
  17.   
  18.     /** 
  19.      * 前置节点 
  20.      */  
  21.     volatile Node prev;  
  22.   
  23.     /** 
  24.      * 后续节点 
  25.      */  
  26.     volatile Node next;  
  27.   
  28.     /** 
  29.      * 节点代表的线程 
  30.      */  
  31.     volatile Thread thread;  
  32.   
  33.     /** 
  34.      *连接到等待condition的下一个节点 
  35.      */  
  36.     Node nextWaiter;  
  37.   
  38. }  

AQS 源码

 

AQS实现了一个同步器的基本结构，下面以独占锁和非独占锁区分来看看 AQS 的几个主要方法：

独占模式

独占获取： tryAcquire 本身不会阻塞线程，如果返回 true 成功就继续，如果返回 false 那么就阻塞线程并加入阻塞队列。

 

Java代码   收藏代码
  1.     public final void acquire(int arg) {  
  2.   
  3.         if (!tryAcquire(arg) &&  
  4.   
  5.             acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//获取失败，则加入等待队列  
  6.   
  7.             selfInterrupt();  
  8.   
  9. }   

 

 

 

独占且可中断模式获取：支持中断取消

Java代码   收藏代码
  1. public final void acquireInterruptibly(int arg) throws InterruptedException {  
  2.   
  3.         if (Thread.interrupted())  
  4.             throw new InterruptedException();  
  5.         if (!tryAcquire(arg))  
  6.   
  7.             doAcquireInterruptibly(arg);  
  8.   
  9.     }   

 

独占且支持超时模式获取： 带有超时时间，如果经过超时时间则会退出。

 

Java代码   收藏代码
  1. public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {  
  2.   
  3.      if (Thread.interrupted())  
  4.   
  5.          throw new InterruptedException();  
  6.   
  7.      return tryAcquire(arg) ||  
  8.   
  9.          doAcquireNanos(arg, nanosTimeout);  

 

 

独占模式释放：释放成功会唤醒后续节点

Java代码   收藏代码
  1. public final boolean release(int arg) {  
  2.     if (tryRelease(arg)) {  
  3.         Node h = head;  
  4.         if (h != null && h.waitStatus != 0)  
  5.             unparkSuccessor(h);  
  6.         return true;  
  7.     }  
  8.     return false;  
  9. }  

 

共享模式

 

共享模式获取

 

Java代码   收藏代码
  1. public final void acquireShared(int arg) {  
  2.   
  3.     if (tryAcquireShared(arg) < 0)  
  4.   
  5.         doAcquireShared(arg);  

 

  可中断模式共享获取

  

Java代码   收藏代码
  1. public final void acquireSharedInterruptibly(int arg) throws InterruptedException {  
  2.         if (Thread.interrupted())  
  3.             throw new InterruptedException();  
  4.         if (tryAcquireShared(arg) < 0)  
  5.             doAcquireSharedInterruptibly(arg);  
  6.     }   

  

共享模式带定时获取

 

Java代码   收藏代码
  1. public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException {  
  2.      if (Thread.interrupted())  
  3.          throw new InterruptedException();  
  4.      return tryAcquireShared(arg) >= 0 ||  
  5.          doAcquireSharedNanos(arg, nanosTimeout);  
  6. }   

 

共享锁释放

  

Java代码   收藏代码
  1. public final boolean releaseShared(int arg) {  
  2.         if (tryReleaseShared(arg)) {  
  3.             doReleaseShared();  
  4.             return true;  
  5.         }  
  6.         return false;  
  7.     }   

 

 

 

注意以上框架只定义了一个同步器的基本结构框架，的基本方法里依赖的 tryAcquire 、 tryRelease 、tryAcquireShared 、 tryReleaseShared 四个方法在 AQS 里没有实现，这四个方法不会涉及线程阻塞，而是由各自不同的使用场景根据情况来定制:

 

 

Java代码   收藏代码
  1. protected boolean tryAcquire(int arg) {  
  2.     throw new UnsupportedOperationException();  
  3. }  
  4. protected boolean tryRelease(int arg) {  
  5.     throw new UnsupportedOperationException();  
  6. }  
  7. protected int tryAcquireShared(int arg) {  
  8.     throw new UnsupportedOperationException();  
  9.   
  10. }  
  11. protected boolean tryReleaseShared(int arg) {  
  12.     throw new UnsupportedOperationException();  
  13. }  

 

从以上源码可以看出AQS实现基本的功能：

AQS虽然实现了acquire，和release方法是可能阻塞的，但是里面调用的tryAcquire和tryRelease是由子类来定制的且是不阻塞的可。以认为同步状态的维护、获取、释放动作是由子类实现的功能，而动作成功与否的后续行为时有AQS框架来实现。所以可以认为同步器实现了一下功能：

1.同步器基本范式、结构

2.状态获取、释放成功或失败的后续行为，如线程的阻塞、唤醒机制

3.线程阻塞队列的维护

 

状态获取、释放动作本身是由子类来定义的。

 

 

还有以下一些私有方法，用于辅助完成以上的功能：

final boolean acquireQueued(final Node node, int arg) ：申请队列

private Node enq(final Node node) : 入队

private Node addWaiter(Node mode) ：以mode创建创建节点，并加入到队列

private void unparkSuccessor(Node node) ： 唤醒节点的后续节点，如果存在的话。

private void doReleaseShared() ：释放共享锁

private void setHeadAndPropagate(Node node, int propagate)：设置头，并且如果是共享模式且propagate大于0，则唤醒后续节点。

private void cancelAcquire(Node node) ：取消正在获取的节点

private static void selfInterrupt() ：自我中断

private final boolean parkAndCheckInterrupt() ： park 并判断线程是否中断

 

 

  • 大小: 29.7 KB
分享到：
| CountDownLatch and CyclicBarrier
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

Magicbox
Global site tag (gtag.js) - Google Analytics