您的位置: 首页  >  文章  >  队列同步器（AQS）详解

队列同步器（AQS）详解

分类: 文章 • 2025-01-25 22:08:22

队列同步器（AQS）

队列同步器AbstractQueuedSynchronizer（以下简称同步器），是用来构建锁或者其他同步组件的基础框架，它使用了一个int成员变量表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作，并发包的作者（Doug Lea）期望它能够成为实现大部分同步需求的基础。

队列同步器的基本结构

同步器依赖内部的同步队列（一个FIFO双向队列）来完成同步状态的管理。同步队列中的节点（Node）用来保存"获取同步状态失败的线程"引用、等待状态以及前驱和后继节点。

同步器包含了两个节点类型的引用，一个指向头节点，而另一个指向尾节点。

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer  
  2.     implements java.io.Serializable {  
  3.     ......  
  4.     private transient volatile Node head;//头节点  
  5.     private transient volatile Node tail;//尾节点  
  6.     private volatile int state;//*同步状态*  
  7.     ......  
  8.     static final class Node {  
  9.         volatile int waitStatus;//等待状态  
  10.         volatile Node prev;//前驱  
  11.         volatile Node next;//后继  
  12.         volatile Thread thread;//线程引用  
  13.         ......  
  14.     }  
  15.     ......  
  16. }   

注：Node类型的prev、next属性以及AbstractQueuedSynchronizer类型的head 、tail属性都设置为volatile，保证可见性。队列同步器（AQS）详解

自定义同步组件的设计思路

同步器的主要使用方式是继承，子类通过继承同步器并实现它的抽象方法来管理同步状态，在抽象方法的实现过程中免不了要对同步状态进行更改，这时就需要使用同步器提供的3个方法（getState()、setState(int newState)和compareAndSetState(int expect,int update)）来进行操作，因为它们能够保证状态的改变是安全的。

子类推荐被定义为自定义同步组件的静态内部类，同步器自身没有实现任何同步接口，它仅仅是定义了若干同步状态获取和释放的方法来供自定义同步组件使用，同步器既可以支持独占式地获取同步状态，也可以支持共享式地获取同步状态，这样就可以方便实现不同类型的同步组件（ReentrantLock、ReentrantReadWriteLock和CountDownLatch等）。

同步器是实现锁（也可以是任意同步组件）的关键，在锁的实现中聚合（组合）同步器，利用同步器实现锁的语义。可以这样理解二者之间的关系：锁是面向使用者的，它定义了使用者与锁交互的接口（比如可以允许两个线程并行访问），隐藏了实现细节；同步器面向的是锁的实现者，它简化了锁的实现方式，屏蔽了同步状态管理、线程的排队、等待与唤醒等底层操作。锁和同步器很好地隔离了使用者和实现者所需关注的领域。

同步器的设计是基于模板方法模式的，也就是说，使用者需要继承同步器并重写指定的方法，随后将同步器组合在自定义同步组件的实现中，并调用同步器提供的模板方法，而这些模板方法将会调用使用者重写的方法。
重写同步器指定的方法时，需要使用同步器提供的如下3个方法来访问或修改同步状态。
getState()：获取当前同步状态。
setState(int newState)：设置当前同步状态。
compareAndSetState(int expect,int update)：使用CAS设置当前状态，该方法能够保证状态设置的原子性。

独占式同步组件的设计

可重写的方法

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. /*Attempts to acquire in exclusive mode. This method should query if the state of the object  
  2. permits it to be acquired in the exclusive mode, and if so to acquire it.*/  
  3. //独占式获取同步状态，实现该方法需要查询当前状态并判断同步状态是否符合预期，然后再进行CAS设置同步状态  
  4. protected boolean tryAcquire(int arg)  
  5.   
  6. /*Attempts to set the state to reflect a release in exclusive mode.*/  
  7. //独占式释放同步状态，等待获取同步状态的线程将有机会获取同步状态  
  8. protected boolean tryRelease(int arg)  
  9.   
  10. /*Returns true if synchronization is held exclusively with respect to the current (calling) thread. */  
  11. //当前同步器是否在独占模式下被线程占用，一般该方法表示是否被当前线程所独占  
  12. protected boolean isHeldExclusively()  

同步器提供的模板方法

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. /*Acquires in exclusive mode, ignoring interrupts.*/  
  2. //独占式获取同步状态，如果当前线程获取同步状态成功，立即返回。否则，将会进入同步队列等待，  
  3. //该方法将会重复调用重写的tryAcquire(int arg)方法  
  4. public final void acquire(int arg)  
  5.   
  6. /*Acquires in exclusive mode, aborting if interrupted.*/  
  7. //与acquire(int arg)基本相同，但是该方法响应中断。  
  8. public final void acquireInterruptibly(int arg)  
  9.   
  10. /* Releases in exclusive mode.  Implemented by unblocking one or more threads if {@link #tryRelease} returns true. 
  11. This method can be used to implement method {@link Lock#unlock}.*/  
  12. //独占式释放同步状态，该方法会在释放同步状态后，将同步队列中第一个节点包含的线程唤醒  
  13. public final boolean release(int arg)  

acquire(int arg)模板方法

通过调用同步器的acquire(int arg)方法可以获取同步状态。该方法对中断不敏感，也就是说，由于线程获取同步状态失败后进入同步队列中，后续对线程进行中断操作时，线程不会从同步队列中移除。

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. public final void acquire(int arg) {//**该方法是模板方法**  
  2.     if (!tryAcquire(arg) &&//先通过tryAcquire获取同步状态  
  3.         acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//获取同步状态失败则生成节点并加入同步队列  
  4.         selfInterrupt();  
  5. }  

独占式同步状态获取流程

主要逻辑：首先调用自定义同步器实现的tryAcquire(int arg)方法，该方法保证线程安全的获取同步状态，如果同步状态获取失败，则构造同步节点（独占式Node.EXCLUSIVE，同一时刻只能有一个线程成功获取同步状态）并通过addWaiter(Node node)方法将该节点加入到同步队列的尾部，最后调用acquireQueued(Node node,int arg)方法，使得该节点以“死循环”的方式获取同步状态。

队列同步器（AQS）详解

将节点加入同步队列

当前线程获取同步状态失败时，同步器会将当前线程、等待状态等信息构造成为一个节点（Node）并将其加入同步队列，同时会阻塞当前线程。

试想一下，当一个线程成功地获取了同步状态（或者锁），其他线程将无法获取到同步状态，转而被构造成为节点并加入到同步队列中，而这个加入队列的过程必须要保证线程安全。

因此，同步器提供了一个基于CAS的设置尾节点的方法：compareAndSetTail(Nodeexpect,Nodeupdate)，它需要传递当前线程“认为”的尾节点和当前节点，只有设置成功后，当前节点才正式与之前的尾节点建立关联。

队列同步器（AQS）详解

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. //将节点加入到同步队列的尾部  
  2. private Node addWaiter(Node mode) {  
  3.       Node node = new Node(Thread.currentThread(), mode);//生成节点（Node）  
  4.       // Try the fast path of enq; backup to full enq on failure  
  5.       //快速尝试在尾部添加  
  6.       Node pred = tail;  
  7.       if (pred != null) {  
  8.           node.prev = pred;//先将当前节点node的前驱指向当前tail  
  9.           if (compareAndSetTail(pred, node)) {//CAS尝试将tail设置为node  
  10.               //如果CAS尝试成功，就说明"设置当前节点node的前驱"与"CAS设置tail"之间没有别的线程设置tail成功  
  11.               //只需要将"之前的tail"的后继节点指向node即可  
  12.               pred.next = node;  
  13.               return node;  
  14.           }  
  15.       }  
  16.       enq(node);//否则，通过死循环来保证节点的正确添加  
  17.       return node;  
  18.   }  
[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. private Node enq(final Node node) {  
  2.     for (;;) {//通过死循环来保证节点的正确添加  
  3.         Node t = tail;  
  4.         if (t == null) { // Must initialize 同步队列为空的情况  
  5.             if (compareAndSetHead(new Node()))  
  6.                 tail = head;  
  7.         } else {  
  8.             node.prev = t;  
  9.             if (compareAndSetTail(t, node)) {//直到CAS成功为止  
  10.                 t.next = node;  
  11.                 return t;//结束循环  
  12.             }  
  13.         }  
  14.     }  
  15. }  

在enq(final Node node)方法中，同步器通过“死循环”来保证节点的正确添加，在“死循环”中只有通过CAS将节点设置成为尾节点之后，当前线程才能从该方法返回，否则，当前线程不断地尝试设置。可以看出，enq(final Node node)方法将并发添加节点的请求通过CAS变得“串行化”了。

串行化的优点

如果通过加锁同步的方式添加节点，线程必须获取锁后才能添加尾节点，那么必然会导致其他线程等待加锁而阻塞，获取锁的线程释放锁后阻塞的线程又会被唤醒，而线程的阻塞和唤醒需要依赖于系统内核完成，因此程序的执行需要从用户态切换到核心态，而这样的切换是非常耗时的操作。如果我们通过”循环CAS“来添加节点的话，所有线程都不会被阻塞，而是不断失败重试，线程不需要进行锁同步，不仅消除了线程阻塞唤醒的开销而且消除了加锁解锁的时间开销。但是循环CAS也有其缺点，循环CAS通过不断尝试来添加节点，如果说CAS操作失败那么将会占用处理器资源。

节点的自旋

节点进入同步队列之后，就进入了一个自旋的过程，每个节点（或者说是线程）都在自省地观察，当条件满足，获取到了同步状态，就可以从这个自旋过程中退出，否则依旧留在这个自旋过程中。

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. final boolean acquireQueued(final Node node, int arg) {  
  2.     boolean failed = true;  
  3.     try {  
  4.         boolean interrupted = false;  
  5.         for (;;) {//无限循环  
  6.             final Node p = node.predecessor();  
  7.             if (p == head && tryAcquire(arg)) {//前驱节点是首节点且获取到了同步状态  
  8.                 setHead(node); //设置首节点  
  9.                 p.next = null; // help GC 断开引用  
  10.                 failed = false;  
  11.                 return interrupted;//从自旋中退出  
  12.             }  
  13.             if (shouldParkAfterFailedAcquire(p, node) &&//获取同步状态失败后判断是否需要阻塞或中断  
  14.                 parkAndCheckInterrupt())//阻塞当前线程  
  15.                 interrupted = true;  
  16.         }  
  17.     } finally {  
  18.         if (failed)  
  19.             cancelAcquire(node);  
  20.     }  
  21. }  
[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. /**Checks and updates status for a node that failed to acquire. 
  2.  * Returns true if thread should block. This is the main signal control in all acquire loops.*/  
  3. private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {  
  4.     int ws = pred.waitStatus;//获取前驱节点的等待状态  
  5.     if (ws == Node.SIGNAL)  
  6.     //SIGNAL状态：前驱节点释放同步状态或者被取消，将会通知后继节点。因此，可以放心的阻塞当前线程，返回true。  
  7.         /* This node has already set status asking a release to signal it, so it can safely park.*/  
  8.         return true;  
  9.     if (ws > 0) {//前驱节点被取消了，跳过前驱节点并重试  
  10.         /* Predecessor was cancelled. Skip over predecessors and indicate retry. */  
  11.         do {  
  12.             node.prev = pred = pred.prev;  
  13.         } while (pred.waitStatus > 0);  
  14.         pred.next = node;  
  15.     } else {//独占模式下，一般情况下这里指前驱节点等待状态为SIGNAL  
  16.         /* waitStatus must be 0 or PROPAGATE.  Indicate that we  need a signal, but don't park yet.  Caller will need to 
  17.          * retry to make sure it cannot acquire before parking. */  
  18.         compareAndSetWaitStatus(pred, ws, Node.SIGNAL);//设置当前节点等待状态为SIGNAL  
  19.     }  
  20.     return false;  
  21. }  
[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. /** Convenience method to park and then check if interrupted 。return {@code true} if interrupted */  
  2. private final boolean parkAndCheckInterrupt() {  
  3.     LockSupport.park(this);//阻塞当前线程  
  4.     return Thread.interrupted();  
  5. }  
队列同步器（AQS）详解

可以看到节点和节点之间在循环检查的过程中基本不相互通信，而是简单地判断自己的前驱是否为头节点，这样就使得节点的释
放规则符合FIFO。并且也便于对过早通知的处理（过早通知是指：前驱节点不是头节点的线程由于中断而被唤醒）。

当同步状态获取成功之后，当前线程从acquire(int arg)方法返回，如果对于锁这种并发组件而言，代表着当前线程获取了锁。

设置首节点

同步队列遵循FIFO，首节点是获取同步状态成功的节点，首节点的线程在释放同步状态时，将会唤醒后续节点，而后续节点将会在获取同步状态成功时将自己设置为首节点。

队列同步器（AQS）详解

设置首节点是由获取同步状态成功的线程来完成的，由于只有一个线程能够成功的获取到同步状态，因此设置头节点的方法并不需要使用CAS来保证，它只需要将首节点设置成为原首节点后继节点，并断开首节点的next引用即可。

释放同步状态

当前线程获取同步状态并执行了相应逻辑之后，就需要释放同步状态，使得后续节点能够继续获取同步状态。通过调用同步器的release(int arg)方法可以释放同步状态，该方法在释放了同步状态之后，会"唤醒"其后继节点（进而使后继节点重新尝试获取同步状态）。

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. public final boolean release(int arg) {  
  2.     if (tryRelease(arg)) {//释放同步状态  
  3.         Node h = head;  
  4.         if (h != null && h.waitStatus != 0)//独占模式下这里表示SIGNAL  
  5.             unparkSuccessor(h);//唤醒后继节点  
  6.         return true;  
  7.     }  
  8.     return false;  
  9. }  
[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. /** Wakes up node's successor, if one exists.*/  
  2. private void unparkSuccessor(Node node) {  
  3.     int ws = node.waitStatus;//获取当前节点等待状态  
  4.     if (ws < 0)  
  5.         compareAndSetWaitStatus(node, ws, 0);//更新等待状态  
  6.   
  7.     /* Thread to unpark is held in successor, which is normally just the next node.   
  8.        But if cancelled or apparently null, 
  9.      * traverse backwards from tail to find the actual non-cancelled successor.*/  
  10.     Node s = node.next;  
  11.     if (s == null || s.waitStatus > 0) {//找到第一个没有被取消的后继节点（等待状态为SIGNAL）  
  12.         s = null;  
  13.         for (Node t = tail; t != null && t != node; t = t.prev)  
  14.             if (t.waitStatus <= 0)  
  15.                 s = t;  
  16.     }  
  17.     if (s != null)  
  18.         LockSupport.unpark(s.thread);//唤醒后继线程  
  19. }  

总结：在获取同步状态时，同步器维护一个同步队列，获取状态失败的线程都会被加入到队列中并在队列中进行自旋；移出队列

（或停止自旋）的条件是前驱节点为头节点且成功获取了同步状态。在释放同步状态时，同步器调用tryRelease(int arg)方法释放同步状态，然后唤醒头节点的后继节点。

独占锁（Mutex）

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. import java.util.Collection;  
  2. import java.util.concurrent.locks.AbstractQueuedSynchronizer;  
  3.    
  4. public class Mutex {  
  5.     // 静态内部类，自定义同步器  
  6.     private static class Sync extends AbstractQueuedSynchronizer {  
  7.         // 是否处于占用状态  
  8.         protected boolean isHeldExclusively() {  
  9.             return getState() == 1;  
  10.         }  
  11.         // 当状态为0的时候获取锁  
  12.         public boolean tryAcquire(int acquires) {  
  13.             if (compareAndSetState(0, 1)) {  
  14.                 setExclusiveOwnerThread(Thread.currentThread());  
  15.                 return true;  
  16.             }  
  17.             return false;  
  18.         }  
  19.         // 释放锁，将状态设置为0  
  20.         protected boolean tryRelease(int releases) {  
  21.             if (getState() == 0)  
  22.                 throw new IllegalMonitorStateException();  
  23.             setExclusiveOwnerThread(null);  
  24.             setState(0);  
  25.             return true;  
  26.         }  
  27.     }  
  28.     // 仅需要将操作代理到Sync上即可  
  29.     private final Sync sync = new Sync();  
  30.       
  31.     //获取等待的线程  
  32.     public Collection<Thread> getQueuedThreads(){  
  33.         return sync.getQueuedThreads();  
  34.     }  
  35.       
  36.         //独占锁的操作接口  
  37.     public void lock() {//获取锁  
  38.         sync.acquire(1);  
  39.     }  
  40.   
  41.     public void unlock() {//释放锁  
  42.         sync.release(1);  
  43.     }  
  44. }  

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. import java.util.Collection;  
  2. import java.util.Random;  
  3.   
  4. public class MutexTestSecond {  
  5.     private static Random r=new Random(47);  
  6.     private static int threadCount=10;  
  7.     private static Mutex mut=new Mutex();  
  8.     private static class Weight implements Runnable{//给苹果称重的任务  
  9.         String name;  
  10.         public Weight(String name){  
  11.             this.name=name;  
  12.         }  
  13.         @Override  
  14.         public void run() {  
  15.             mut.lock();  
  16.             System.out.println(name+"放苹果！");  
  17.             System.out.println(name+"重量："+(r.nextInt(10)+3));  
  18.             System.out.println(name+"取苹果！");  
  19.             printQueuedThreads(mut.getQueuedThreads());  
  20.             mut.unlock();  
  21.         }   
  22.     }  
  23.     private static void printQueuedThreads(Collection<Thread> threads){  
  24.         System.out.print("等待队列中的线程：");  
  25.         for(Thread t:threads){  
  26.             System.out.print(t.getName()+" ");  
  27.         }  
  28.         System.out.println();  
  29.     }  
  30.     public static void main(String[] args) {  
  31.          Thread[] threads=new Thread[threadCount];  
  32.          for(int i=0;i<threadCount;i++){  
  33.              threads[i]=new Thread(new Weight("Weight-"+i),"Thread-"+i);  
  34.          }  
  35.          for(int i=0;i<threadCount;i++){  
  36.              threads[i].start();  
  37.          }  
  38.     }  
  39. }  

输出：

Weight-0放苹果！
Weight-0重量：11
Weight-0取苹果！
等待队列中的线程：Thread-3 Thread-2 Thread-1 
Weight-6放苹果！
Weight-6重量：8
Weight-6取苹果！
等待队列中的线程：Thread-8 Thread-7 Thread-5 Thread-4 Thread-3 Thread-2 Thread-1 
Weight-1放苹果！
Weight-1重量：6
Weight-1取苹果！
等待队列中的线程：Thread-9 Thread-8 Thread-7 Thread-5 Thread-4 Thread-3 Thread-2 
Weight-2放苹果！
Weight-2重量：4
Weight-2取苹果！
等待队列中的线程：Thread-9 Thread-8 Thread-7 Thread-5 Thread-4 Thread-3 
Weight-3放苹果！
Weight-3重量：4
Weight-3取苹果！
等待队列中的线程：Thread-9 Thread-8 Thread-7 Thread-5 Thread-4 
Weight-4放苹果！
Weight-4重量：12
Weight-4取苹果！
等待队列中的线程：Thread-9 Thread-8 Thread-7 Thread-5 
Weight-5放苹果！
Weight-5重量：11
Weight-5取苹果！
等待队列中的线程：Thread-9 Thread-8 Thread-7 
Weight-7放苹果！
Weight-7重量：3
Weight-7取苹果！
等待队列中的线程：Thread-9 Thread-8 
Weight-8放苹果！
Weight-8重量：5
Weight-8取苹果！
等待队列中的线程：Thread-9 
Weight-9放苹果！
Weight-9重量：10
Weight-9取苹果！
等待队列中的线程：

从输出中可以看出，我们的独占锁Mutex，保证了秤的独占使用。

重入锁

重入锁ReentrantLock，顾名思义，就是支持重进入的锁，它表示该锁能够支持一个线程对资源的重复加锁。除此之外，该锁的还支持获取锁时的公平和非公平性选择。

对于独占锁（Mutex），考虑如下场景：当一个线程调用Mutex的lock()方法获取锁之后，如果再次调用lock()方法，则该线程将会被自己所阻塞，原因是Mutex在实现tryAcquire(int acquires)方法时没有考虑占有锁的线程再次获取锁的场景，而在调用tryAcquire(int acquires)方法时返回了false，导致该线程被阻塞。简单地说，Mutex是一个不支持重进入的锁。

synchronized关键字隐式的支持重进入，比如一个synchronized修饰的递归方法，在方法执行时，执行线程在获取了锁之后仍能连续多次地获得该锁，而不像Mutex由于获取了锁，而在下一次获取锁时出现阻塞自己的情况。
ReentrantLock虽然没能像synchronized关键字一样支持隐式的重进入，但是在调用lock()方法时，已经获取到锁的线程，能够再次调用lock()方法获取锁而不被阻塞。

可重入的实现

重进入是指任意线程在获取到锁之后能够再次获取该锁而不会被锁所阻塞，该特性的实现需要解决以下两个问题。
1）线程再次获取锁。锁需要去识别获取锁的线程是否为当前占据锁的线程，如果是，则再次成功获取。
2）锁的最终释放。线程重复n次获取了锁，随后在第n次释放该锁后，其他线程能够获取到该锁。锁的最终释放要求锁对于获取进行计数自增，计数表示当前锁被重复获取的次数，而锁被释放时，计数自减，当计数等于0时表示锁已经成功释放。

ReentrantLock是通过组合自定义同步器来实现锁的获取与释放。我们以非公平锁为例：

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. public class ReentrantLock implements Lock, java.io.Serializable {  
  2.     private final Sync sync;  
  3.   
  4.     ......  
  5.     abstract static class Sync extends AbstractQueuedSynchronizer {  
  6.         private static final long serialVersionUID = -5179523762034025860L;  
  7.   
  8.         abstract void lock();//抽象方法  
  9.   
  10.         final boolean nonfairTryAcquire(int acquires) {//非公平的获取锁  
  11.             final Thread current = Thread.currentThread();  
  12.             int c = getState();  
  13.             if (c == 0) {//首次获取同步状态  
  14.                 if (compareAndSetState(0, acquires)) {//只要设置成功就获取到锁  
  15.                     setExclusiveOwnerThread(current);  
  16.                     return true;  
  17.                 }  
  18.             }  
  19.             else if (current == getExclusiveOwnerThread()) {//再次获取同步状态(可重入的关键)  
  20.                 //如果是获取锁的线程再次请求，则将同步状态值进行增加并返回true，表示获取同步状态成功。  
  21.                 int nextc = c + acquires;  
  22.                 if (nextc < 0) // overflow  
  23.                     throw new Error("Maximum lock count exceeded");  
  24.                 setState(nextc);  
  25.                 return true;  
  26.             }  
  27.             return false;  
  28.         }  
  29.   
  30.         protected final boolean tryRelease(int releases) {  
  31.             int c = getState() - releases;  
  32.             if (Thread.currentThread() != getExclusiveOwnerThread())  
  33.                 throw new IllegalMonitorStateException();  
  34.             boolean free = false;  
  35.             if (c == 0) {//当同步状态为0时，将占有线程设置为null  
  36.                 free = true;  
  37.                 setExclusiveOwnerThread(null);  
  38.             }  
  39.             setState(c);//更新同步状态  
  40.             return free;  
  41.         }  
  42.         ......  
  43.     }  
  44.     static final class NonfairSync extends Sync {  
  45.         private static final long serialVersionUID = 7316153563782823691L;  
  46.   
  47.         /**Performs lock.  Try immediate barge, backing up to normal acquire on failure. */  
  48.         final void lock() {  
  49.             if (compareAndSetState(0, 1))//首次获取锁成功  
  50.                 setExclusiveOwnerThread(Thread.currentThread());  
  51.             else  
  52.                 acquire(1);//申请加锁  
  53.         }  
  54.   
  55.         protected final boolean tryAcquire(int acquires) {  
  56.             return nonfairTryAcquire(acquires);//非公平获取锁  
  57.         }  
  58.     }  
  59.     ......  
  60.     public void lock() {  
  61.         sync.lock();  
  62.     }  
  63.     public void unlock() {  
  64.         sync.release(1);  
  65.     }  
  66.     ......  
  67. }  

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. import java.util.Random;  
  2. import java.util.concurrent.locks.ReentrantLock;  
  3.    
  4. public class ReentrantLockTest {  
  5.     private static Random r=new Random(47);  
  6.     private static int threadCount=10;  
  7.     private static ReentrantLock mut=new ReentrantLock();  
  8.     private static class Weight implements Runnable{//给苹果称重的任务  
  9.         String name;  
  10.         public Weight(String name){  
  11.             this.name=name;  
  12.         }  
  13.         @Override  
  14.         public void run() {   
  15.             mut.lock();  
  16.             System.out.println(name+"放苹果！");  
  17.             System.out.println(name+"重量："+(r.nextInt(10)+3));  
  18.             System.out.println(name+"取苹果！");   
  19.             if(r.nextInt()%2==0){run();}//递归调用  
  20.             mut.unlock();  
  21.         }   
  22.     }   
  23.     public static void main(String[] args) throws InterruptedException {  
  24.          Thread[] threads=new Thread[threadCount];  
  25.          for(int i=0;i<threadCount;i++){  
  26.              threads[i]=new Thread(new Weight("Weight-"+i),"Thread-"+i);  
  27.          }  
  28.          for(int i=0;i<threadCount;i++){  
  29.              threads[i].start();  
  30.              Thread.sleep(10);  
  31.          }  
  32.     }  
  33. }  
输出：

Weight-0放苹果！
Weight-0重量：11
Weight-0取苹果！
Weight-0放苹果！
Weight-0重量：6
Weight-0取苹果！
Weight-0放苹果！
Weight-0重量：4
Weight-0取苹果！
Weight-1放苹果！
Weight-1重量：11
Weight-1取苹果！
Weight-2放苹果！
Weight-2重量：5
Weight-2取苹果！
Weight-2放苹果！
Weight-2重量：11
Weight-2取苹果！
Weight-2放苹果！
Weight-2重量：4
Weight-2取苹果！
Weight-3放苹果！
Weight-3重量：12
Weight-3取苹果！
Weight-3放苹果！
Weight-3重量：11
Weight-3取苹果！
Weight-3放苹果！
Weight-3重量：3
Weight-3取苹果！
Weight-3放苹果！
Weight-3重量：9
Weight-3取苹果！
Weight-5放苹果！
Weight-5重量：4
Weight-5取苹果！
Weight-7放苹果！
Weight-7重量：7
Weight-7取苹果！
Weight-8放苹果！
Weight-8重量：9
Weight-8取苹果！
Weight-6放苹果！
Weight-6重量：3
Weight-6取苹果！
Weight-4放苹果！
Weight-4重量：7
Weight-4取苹果！
Weight-4放苹果！
Weight-4重量：3
Weight-4取苹果！
Weight-9放苹果！
Weight-9重量：5
Weight-9取苹果！
Weight-9放苹果！
Weight-9重量：6
Weight-9取苹果！
Weight-9放苹果！
Weight-9重量：7
Weight-9取苹果！

从输出中，可以看出可重入特性。如果，我们将可重入锁换成独占锁Mutex程序将会阻塞，不具有可重入性。

此外，我们还发现，线程的执行是乱序的（从线程名称的角度看），即与start()方法调用顺序不一致。这是为什么呢？

原来重入锁ReentrantLock默认采用非公平实现？那好，我们将可重入锁设置为公平锁：

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. private static ReentrantLock mut=new ReentrantLock(true);//设置为公平锁  

输出：

Weight-0放苹果！
Weight-0重量：11
Weight-0取苹果！
Weight-0放苹果！
Weight-0重量：6
Weight-0取苹果！
Weight-0放苹果！
Weight-0重量：4
Weight-0取苹果！
Weight-1放苹果！
Weight-1重量：11
Weight-1取苹果！
Weight-2放苹果！
Weight-2重量：5
Weight-2取苹果！
Weight-2放苹果！
Weight-2重量：11
Weight-2取苹果！
Weight-2放苹果！
Weight-2重量：4
Weight-2取苹果！
Weight-3放苹果！
Weight-3重量：12
Weight-3取苹果！
Weight-3放苹果！
Weight-3重量：11
Weight-3取苹果！
Weight-3放苹果！
Weight-3重量：3
Weight-3取苹果！
Weight-3放苹果！
Weight-3重量：9
Weight-3取苹果！
Weight-7放苹果！
Weight-7重量：4
Weight-7取苹果！
Weight-6放苹果！
Weight-6重量：7
Weight-6取苹果！
Weight-4放苹果！
Weight-4重量：9
Weight-4取苹果！
Weight-5放苹果！
Weight-5重量：3
Weight-5取苹果！
Weight-8放苹果！
Weight-8重量：7
Weight-8取苹果！
Weight-8放苹果！
Weight-8重量：3
Weight-8取苹果！
Weight-9放苹果！
Weight-9重量：5
Weight-9取苹果！
Weight-9放苹果！
Weight-9重量：6
Weight-9取苹果！
Weight-9放苹果！
Weight-9重量：7
Weight-9取苹果！

从输出中我们看到，线程的执行顺序与对应start()方法被调用的顺序依然不一样，说好的公平锁呢？

原因分析：start()语句调用的顺序与线程进入Runnable状态的顺序不一定一致，也就是说先调用start()语句所对应的线程不一定先进入Runnable状态，即使先进入Runnable状态也不一定先分得处理器开始执行。

公平性与否是针对获取锁而言的，如果一个锁是公平的，那么锁的获取顺序就应该符合请求的绝对时间顺序，也就是FIFO。

个人理解，如有偏颇，还望指正！！

非公平性的实现

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. static final class NonfairSync extends Sync {  
  2.     private static final long serialVersionUID = 7316153563782823691L;  
  3.   
  4.     /**Performs lock.  Try immediate barge, backing up to normal acquire on failure.*/  
  5.     final void lock() {  
  6.         if (compareAndSetState(0, 1))//只要CAS更新同步状态成功就获取到锁。  
  7.             setExclusiveOwnerThread(Thread.currentThread());  
  8.         else  
  9.             acquire(1);  
  10.     }  
  11.   
  12.     protected final boolean tryAcquire(int acquires) {  
  13.         return nonfairTryAcquire(acquires);  
  14.     }  
  15. }  
非公平性实例，如果Thread-1拥有锁，Thread-2和Thread-3在同步队列中，当Thread-1释放锁后会唤醒Thread-2，但是如果此时Thread-1重新申请锁，可能依然是Thread-1获取到锁。甚至这时候Thread-4也申请锁，Thread-4也可能比Thread-2先获取锁。

非公平锁可能使得线程“饥饿”。当一个线程请求锁时，只要获取了同步状态即成功获取锁。在这个前提下，刚释放锁的线程再次获取同步状态的几率会非常大，使得其他线程只能在同步队列中等待。

公平性的实现

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. static final class FairSync extends Sync {  
  2.     private static final long serialVersionUID = -3000897897090466540L;  
  3.   
  4.     final void lock() {  
  5.         acquire(1);  
  6.     }  
  7.   
  8.     /**  Fair version of tryAcquire.  Don't grant access unless recursive call or no waiters or is first.*/  
  9.     protected final boolean tryAcquire(int acquires) {  
  10.         final Thread current = Thread.currentThread();  
  11.         int c = getState();  
  12.         //这里没有一进来就直接进行CAS操作  
  13.         if (c == 0) {  
  14.             if (!hasQueuedPredecessors() &&<span><span class="comment">//增加是否有前驱线程的判断</span><span>，从而保证公平性</span></span>  
  15.                 compareAndSetState(0, acquires)) {  
  16.                 setExclusiveOwnerThread(current);  
  17.                 return true;  
  18.             }  
  19.         }  
  20.         else if (current == getExclusiveOwnerThread()) {  
  21.             int nextc = c + acquires;  
  22.             if (nextc < 0)  
  23.                 throw new Error("Maximum lock count exceeded");  
  24.             setState(nextc);  
  25.             return true;  
  26.         }  
  27.         return false;  
  28.     }  
  29. }  

公平锁与非公平锁的比较

公平性与否是针对获取锁而言的，如果一个锁是公平的，那么锁的获取顺序就应该符合请求的绝对时间顺序，也就是FIFO。

公平性锁每次都是从同步队列中的第一个节点获取到锁，而非公平性锁出现了一个线程连续获取锁的情况。
非公平性锁可能使线程“饥饿”，当一个线程请求锁时，只要获取了同步状态即成功获取锁。在这个前提下，刚释放锁的线程再次获取同步状态的几率会非常大，使得其他线程只能在同步队列中等待。
非公平锁可能使线程“饥饿”，为什么它又被设定成默认的实现呢？非公平性锁模式下线程上下文切换的次数少，因此其性能开销更小。公平性锁保证了锁的获取按照FIFO原则，而代价是进行大量的线程切换。非公平性锁虽然可能造成线程“饥饿”，但极少的线程切换，保证了其更大的吞吐量。

共享式同步组件设计

可重写的方法

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1. /**Attempts to acquire in shared mode. This method should query if the state of the object 
  2. permits it to be acquired in the shared mode, and if so to acquire it.*/  
  3. //共享式获取同步状态，返回大于等于0的值，表示获取成功，反之获取失败  
  4. protected int tryAcquireShared(int arg)  
  5.   
  6.   
  7. /**Attempts to set the state to reflect a release in shared mode.*/  
  8. //共享式释放同步状态  
  9. protected boolean tryReleaseShared(int arg)  

同步器提供的模板方法

[java] view plain copy
 队列同步器（AQS）详解队列同步器（AQS）详解
  1.     /**Acquires in shared mode, ignoring interrupts.*/  
  2.     //共享式获取同步状态，如果当前线程未获取到同步状态，将会进入同步队列等待  
  3.     //与独占式获取的主要区别是在同一时刻可以有多个线程获取到同步状态  
  4.     public final void acquireShared(int arg)  
  5.   
  6. <span style="font-size:14px;"><span style="font-size:14px;"></span></span><pre name="code" class="java">    /**Acquires in exclusive mode, aborting if interrupted.*/  
  7.     //该方法可以响应中断  
  8.     public final void acquireInterruptibly(int arg)  

/**Releases in shared mode. Implemented by unblocking one or more * threads if {@link #tryReleaseShared} returns true.*/ //共享式释放同步状态 public final boolean releaseShared(int arg) 












共享式获取与独占式获取最主要的区别在于同一时刻能否有多个线程同时获取到同步状态。以文件的读写为例，如果一个程序在对文件进行读操作，那么这一时刻对于该文件的写操作均被阻塞，而读操作能够同时进行。写操作要求对资源的独占式访问，而读操作可以是共享式访问。







队列同步器（AQS）详解








获取同步状态








调用同步器的acquireShared(int arg)方法可以共享式地获取同步状态。







[java] view
plain copy




 队列同步器（AQS）详解队列同步器（AQS）详解





    

  1. 
public final void acquireShared(int arg) {  

    2. 

  2. 
    if (tryAcquireShared(arg) < 0)  

    3. 

  3. 
        doAcquireShared(arg);  

    4. 

  4. 
}  

    5. 






在acquireShared(int arg)方法中，同步器调用tryAcquireShared(int arg)方法尝试获取同步状态，tryAcquireShared(int arg)方法返回值为int类型，当返回值大于等于0时，表示能够获取到同步状态。因此，在共享式获取的自旋过程中，成功获取到同步状态并退出自旋的条件就是tryAcquireShared(int arg)方法返回值大于等于0。



在doAcquireShared(int arg)方法的自旋过程中，如果当前节点的前驱为头节点时，尝试获取同步状态，如果返回值大于等于0，表示该次获取同步状态成功并从自旋过程中退出。



释放同步状态





与独占式一样，共享式获取也需要释放同步状态，通过调用releaseShared(int arg)方法可以释放同步状态。







[java] view
plain copy




 队列同步器（AQS）详解队列同步器（AQS）详解





    

  1. 
public final boolean releaseShared(int arg) {  

    2. 

  2. 
    if (tryReleaseShared(arg)) {  

    3. 

  3. 
        doReleaseShared();  

    4. 

  4. 
        return true;  

    5. 

  5. 
    }  

    6. 

  6. 
    return false;  

    7. 

  7. 
}  

    8. 






该方法在释放同步状态之后，将会唤醒后续处于等待状态的节点。对于能够支持多个线程同时访问的并发组件（比Semaphore），它和独占式主要区别在于tryReleaseShared(int arg)方法必须确保同步状态（或者资源数）线程安全释放，一般是通过循环和CAS来保证的，因为释放同步状态的操作会同时来自多个线程。








CountDownLatch
















[java] view
plain copy




 队列同步器（AQS）详解队列同步器（AQS）详解





    

  1. 
public class CountDownLatch {    

    2. 

  2. 
    /**Synchronization control For CountDownLatch. Uses AQS state to represent count.*/    

    3. 

  3. 
    private static final class Sync extends AbstractQueuedSynchronizer {    

    4. 

  4. 
        private static final long serialVersionUID = 4982264981922014374L;    

    5. 

  5. 
    

    6. 

  6. 
        Sync(int count) {    

    7. 

  7. 
            setState(count);//初始化同步状态    

    8. 

  8. 
        }    

    9. 

  9. 
    

    10. 

  10. 
        int getCount() {    

    11. 

  11. 
            return getState();    

    12. 

  12. 
        }    

    13. 

  13. 
    

    14. 

  14. 
        protected int tryAcquireShared(int acquires) { //同步状态为0才返回成功  

    15. 

  15. 
            return (getState() == 0) ? 1 : -1;    

    16. 

  16. 
        }    

    17. 

  17. 
    

    18. 

  18. 
        protected boolean tryReleaseShared(int releases) {//减少同步状态  

    19. 

  19. 
            // Decrement count; signal when transition to zero    

    20. 

  20. 
            for (;;) {  //这里通过循环CAS来释放同步状态，从而保证线程安全性  

    21. 

  21. 
                int c = getState();    

    22. 

  22. 
                if (c == 0)    

    23. 

  23. 
                    return false;    

    24. 

  24. 
                int nextc = c-1;    

    25. 

  25. 
                if (compareAndSetState(c, nextc))    

    26. 

  26. 
                    return nextc == 0;    

    27. 

  27. 
            }    

    28. 

  28. 
        }    

    29. 

  29. 
    }    

    30. 

  30. 
    

    31. 

  31. 
    private final Sync sync;//组合一个同步器（AQS）    

    32. 

  32. 
    

    33. 

  33. 
    public CountDownLatch(int count) {    

    34. 

  34. 
        if (count < 0) throw new IllegalArgumentException("count < 0");    

    35. 

  35. 
        this.sync = new Sync(count);//初始化同步状态    

    36. 

  36. 
    }    

    37. 

  37. 
  

    38. 

  38. 
    /*Causes the current thread to wait until the latch has counted down to  

    39. 

  39. 
     * zero, unless the thread is {@linkplain Thread#interrupt interrupted}.*/    

    40. 

  40. 
    public void await() throws InterruptedException {    

    41. 

  41. 
        sync.acquireSharedInterruptibly(1);//当同步状态为0时，acquireShared(1)才返回  

    42. 

  42. 
    }    

    43. 

  43. 
    

    44. 

  44. 
    public boolean await(long timeout, TimeUnit unit)    

    45. 

  45. 
        throws InterruptedException {    

    46. 

  46. 
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));    

    47. 

  47. 
    }    

    48. 

  48. 
    public void countDown() {    

    49. 

  49. 
        sync.releaseShared(1);//释放同步状态    

    50. 

  50. 
    }    

    51. 

  51. 
    

    52. 

  52. 
    public long getCount() {    

    53. 

  53. 
        return sync.getCount();    

    54. 

  54. 
    }    

    55. 

  55. 
    

    56. 

  56. 
    public String toString() {    

    57. 

  57. 
        return super.toString() + "[Count = " + sync.getCount() + "]";    

    58. 

  58. 
    }    

    59. 

  59. 
}    

    60. 










独占式超时获取同步状态



队列同步器（AQS）详解





Lock接口







锁是用来控制多个线程访问共享资源的方式，一般来说，一个锁能够防止多个线程同时访问共享资源（但是有些锁可以允许多个线程并发的访问共享资源，比如读写锁）。



在Lock接口出现之前，Java程序是靠synchronized关键字实现锁功能的，而Java
SE 5之后，并发包中新增了Lock接口（以及相关实现类）用来实现锁功能，它提供了与synchronized关键字类似的同步功能，只是在使用时需要显式地获取和释放锁。虽然它缺少了（通过synchronized块或者方法所提供的）隐式获取释放锁的便捷性，但是却拥有了锁获取与释放的可操作性、可中断的获取锁以及超时获取锁等多种synchronized关键字所不具备的同步特性。

使用synchronized关键字将会隐式地获取锁，但是它将锁的获取和释放固化了，也就是先获取再释放。当然，这种方式简化了同步的管理，可是扩展性没有显示的锁获取和释放来的好。例如，针对一个场景，手把手进行锁获取和释放，先获得锁A，然后再获取锁B，当锁B获得后，释放锁A同时获取锁C，当锁C获得后，再释放B同时获取锁D，以此类推。这种场景下，synchronized关键字就不那么容易实现了，而使用Lock却容易许多。



Lock的使用方式








[java] view
plain copy




 队列同步器（AQS）详解队列同步器（AQS）详解





    

  1. 
Lock lock = new ReentrantLock();  

    2. 

  2. 
lock.lock();  

    3. 

  3. 
try {  

    4. 

  4. 
。。。。。。  

    5. 

  5. 
} finally {  

    6. 

  6. 
     lock.unlock();  

    7. 

  7. 
}  

    8. 




在finally块中释放锁，目的是保证在获取到锁之后，最终能够被释放。

不要将获取锁的过程写在try块中，因为如果在获取锁（自定义锁的实现）时发生了异常，异常抛出的同时，也会导致锁无故释放。



Lock接口提供的新特性




队列同步器（AQS）详解



Lock接口








[java] view
plain copy




 队列同步器（AQS）详解队列同步器（AQS）详解





    

  1. 
public interface Lock {  

    2. 

  2. 
    //获取锁，调用该方法将会获取锁，当锁获取后，从该方法返回  

    3. 

  3. 
    void lock();  

    4. 

  4. 
    //可中断地获取锁，和lock()方法的不同之处在于该方法会响应中断，即在锁的获取过程中可以中断当前线程   

    5. 

  5. 
    void lockInterruptibly() throws InterruptedException;  

    6. 

  6. 
    //尝试非阻塞的获取锁，调用该方法后会立刻返回，如果能够获取则返回true,否则返回false   

    7. 

  7. 
    boolean tryLock();  

    8. 

  8. 
    //超时地获取锁 1、当前线程在超时时间内成功获取锁。2、当前线程在超时时间内被中断。3、超时时间结束返回false。   

    9. 

  9. 
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;  

    10. 

  10. 
    //释放锁   

    11. 

  11. 
    void unlock();  

    12. 

  12. 
    //获取等待通知组件   

    13. 

  13. 
    Condition newCondition();  

    14. 

  14. 
}  

    15. 






Lock接口的实现




Lock接口的实现基本都是通过组合了一个队列同步器（AbstractQueuedSynchronizer）的子类来完成线程访问控制的。



例如，ReentrantLock（重入锁）。

说在最后：关于可响应中断和超时等待特性，文中基本略过，详情可参看《Java并发编程的艺术》和JDK源码。

参考：



JDK 1.7源码



《Java并发编程的艺术》



《Java并发编程实践》
															

																					

								

									相关推荐