AVA并发编程学习笔记之AQS简介

AVA并发编程学习笔记之AQS简介
分类： JAVA并发编程 2012-05-10 21:19 815人阅读 评论(1) 收藏 举报
java编程thread框架semaphorejvm
目录(?)[+]
1、引言
JAVA内置的锁（使用同步方法和同步块）一直以来备受关注，其优势是可以花最小的空间开销创建锁（因为每个JAVA对象或者类都可以作为锁使用）和最少的时间开销获得锁（单线程可以在最短时间内获得锁）。线程同步越来越多地被用在多处理器上，特别是在高并发的情况下，然而，JVM内置锁表现一般，而且不支持任何公平策略。从JAVA 5开始在java.util.concurrent包中引入了有别于Synchronized的同步框架。
下面谈谈它的设计思路：
设计一个同步器至少应该具以下有两种操作：一个获取方法，如果当前状态不允许，将一直阻塞这个线程；一个释放方法，修改状态，让其他线程有运行的机会。并发包中并没有为同步器提供一个统一的API，获取和释放方法在不同的类中的名称不同，比如获取方法有：Lock.lock,Semaphore.acquire, CountDownLatch.await和FutureTask.get.这些方法一般都重载有多种版本：阻塞与非阻塞版本、支持超时、支持中断。
java.util.concurrent包中有很多同步类，比如互斥锁、读写锁、信号量等，这些同步类几乎都可以用不同方式来实现，但是如果这样做，那么这样的项目充其量只能算一个二流工程。JSR166并没有生搬硬套，而是建立了一个同步中心类AbstractQueuedSynchronizer(简称：AQS)的框架，其中提供了大量的同步操作，而且用户还可以在此类的基础上自定义自己的同步类。其设计目标主要有两点：
1、提高可扩展性，用户可以自定义自己的同步类
2、最大限度地提高吞吐量，提供自定义公平策略

2、设计和实现
同步器的设计比较直接，前面提到包含获取和释放两个操作：
获取操作过程如下：
while (synchronization state does not allow acquire) {
    enqueue current thread if not already queued;
    possibly block current thread;
}
dequeue current thread if it was queued;
释放操作：
update synchronization state;
if (state may permit a blocked thread to acquire)
    unblock one or more queued threads;
要满足以上两个操作，需要以下3点来支持：
1、原子操作同步状态；
2、阻塞或者唤醒一个线程；
3、内部应该维护一个队列。

2.1同步状态
AQS用的是一个32位的整型来表示同步状态的，可以通过以下几个方法来设置和修改这个状态字段：getState(),setState(),compareAndSetState().这些方法都需要java.util.concurrent.atomic包的支持，采用CAS操作.将state设置为32位整型是一个务实的决定，虽然JSR166提供了64位版本的原子操作，但它还是使用对象内部锁来实现的，如果采用64位的state会导致同步器表现不良好。32位同步器满足大部分应用，如果确实需要64位的状态，可以使用AbstractQueuedLongSynchronizer类.AQS是一个抽象类，如果它的实现类想要想要拥有对获取和释放的控制权，那它必须实现tryAcquire和tryRelease两个方法。
[html] view plaincopyprint?
protected final int getState() { 
    return state; 
} 
 
 
protected final void setState(int newState) { 
    state = newState; 
} 
 
 
protected final boolean compareAndSetState(int expect, int update) { 
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update); 
} 
 
 
protected boolean tryAcquire(int arg) { 
    throw new UnsupportedOperationException(); 
} 
 
 
protected boolean tryRelease(int arg) { 
    throw new UnsupportedOperationException(); 
} 

2.2阻塞
JSR166以前还没有好的阻塞和解除阻塞线程的API可以使用！只有Thread.suspend 和 Thread.resume，但这两个方法已经被废弃了，原因是有可能导致死锁。如果一个线程拥有监视器然后调用 Thread.suspend 使自已阻塞，另一个线程试图调用Thread.resume去唤醒它，那么这个线程去获取监视器时即出现死锁。直到后来出现的LockSupport解决了这个问题，LockSupport.park可以阻塞一个线程，LockSupport.unpack可以解除阻塞，调用一次park,然后调用多次unpack只会唤醒一个线程，阻塞针对线程而不是针对同步器。特别的，如果一个线程在一个新的同步器上调用pack方法有可能立即返回，因为可能有剩余的unpack存在。虽然调用多次unpack是想彻底清除阻塞状态，但这显得很笨拙，而且不划算，更有效的做法是在多次park的时候才多次unpark.

2.3队列
同步框架最重要的是要有一个同步队列，在这里被严格限制为FIFO队列，因此这个同步框架不支持基于优先级的同步策略。同步队列采用非阻塞队列毋庸置疑，当时非阻塞队列只有两个可供选择CLH队列锁和MCS队列锁.原始的CLH Lock仅仅使用自旋锁，但是相对于MSC Lock它更容易处理cancel和timeout，所以选择了CLH Lock。
CLH队列锁的优点是：进出队快，无锁，畅通无阻（即使在有竞争的情况下，总有一个线程总是能够很快插入到队尾）；检查是否有线程在等待也是很容易的（只需要检查头尾指针是否相同）。最后设计出来的变种CLH Lock和原始的CLH Lock有较大的差别：
1、为了可以处理timeout和cancel操作，每个node维护一个指向前驱的指针。如果一个node的前驱被cancel，这个node可以前向移动使用前驱的状态字段。
2、第二个变动是在每个node里使用一个状态字段去控制阻塞，而不是自旋。一个排队的线程调用acquire，只有在通过了子类实现的tryAcquire才能返回，确保只有队头线程才允许调用tryAcquire。
3、另外还有一些微小的改动：head结点使用的是傀儡结点。
变种的CLH队列如下图所示：


2.4条件队列
同步框架提供了一个ConditionObject，一般和Lock接口配合来支持互斥模型，它提供类似JVM同步器的操作。条件对象可以和其他同步器有效的整合，它修复了JVM内置同步器的不足：一个锁可以有多个条件。条件结点内部也有一个状态字段，条件结点是通过nextWaiter指针串起来的一个独立的队列。条件队列中的线程在获取锁之前，必须先被transfer到同步队列中去。transfer先断开条件队列的第一个结点，然后插入到同步队列中，这个新插入到同步队列中的结点和同步队列中的结点一起排队等待获取锁。

3、用法
AbstractQueuedSynchronizer是一个采用模板方法模式实现的同步器基类，子类只需要实现获取和释放方法。子类一般不直接用于同步控制，而是采用代理模式。因为获取和释放方法一般是私有的，实现细节不必暴露出来，所以常用委派的方法来使用同步器类：在一个类的内部申请一个私有的AQS的子类，委派它的所有同步方法。
[java] view plaincopyprint?
class Mutex { 
    class Sync extends AbstractQueuedSynchronizer { 
        public boolean tryAcquire(int ignore) { 
            return compareAndSetState(0, 1); 
        } 
        public boolean tryRelease(int ignore) { 
            setState(0);  
            return true; 
        } 
    } 
 
    private final Sync sync = new Sync(); 
 
    public void lock() {  
        sync.acquire(0);  
    } 
 
    public void unlock() {  
        sync.release(0);  
    } 
} 
java.util.concurrent包中的所有同步工具类都依赖于AQS，其类型程序结构图如下：



AbstractQueuedSynchronizer类还提供了其他一些同步控制方法，包括超时和中断版的获取方法，还集成了独占模式的同步器，如acquireShared，tryReleaseShared等方法。

3.1控制公平
虽然这个队列被设计为FIFO，但并不意味着这个同步器一定是公平的，前面谈到，在tryAcquire检查之后再排队。因此，新线程完全可以偷偷排在第一个线程前面。之所以不采用FIFO，有时候是想获得更高的吞吐量，为了减少等待时间，新到的线程与队列头部的线程一起公平竞争，如果新来的线程比队头的线程快，那么这个新来的线程就获取锁。队头线程失去竞争会再次阻塞，它的继任也将会被阻塞，但这样能避免饥饿。

如果需要绝对公平，那很简单，只需要在tryAcquire方法，不在队头返回false即可。检查是否在队头可以使用getFirstQueuedThread方法。有一情况是，队列是空的，同时有多个线程一拥而入，谁先抢到锁就谁运行，这其实与公平并不冲突，是对公平的补充。

3.2同步器
JAVA并发框架是如何使用AQS的：
ReentrantLock类使用同步状态来代表持有锁的数量，当一个锁被获得，会记录获取该锁的线程身份，如果一个非当前线程试图释放锁是不合法的。该类也使用了ConditionObject类，和一些监视和检查方法。该类支持公平与非公平两种模式，是通过AQS的两个子类来实现的。
ReentrantReadWriteLock类将32位的state分成高位和低位，16位用于写锁计数，其余16位用于读锁计数。
Semaphore类使用同步状态保持当前计数，acquireShared减少计数，tryRelease的增加计数，如果state是正数就唤醒线程。
CountDownLatch类使用同步状态代表计数。所有线程都获得锁时，状态为0，就唤醒。
当然用户可以定义自己的应用程序同步器。例如：事件，集中管理的锁，基于树的障碍等。