队列同步器AbstractQueuedSynchronizer

　　AQS是用来构建锁或者其它同步组件的基础框架，它使用一个int变量来表示同步状态，通过内置的FIFO队列来完成获取线程的排队工作，concurrent包的作者Doug Lea期望它能称为实现大部分同步需求的基础。

　　同步器的使用方式是继承，子类通过继承AQS并实现它的相关方法来管理同步状态，在子类方法的实现中不可避免的要对状态进行更改，AQS提供了3个方法：getState()、setState(int newState)跟compareAndSetState(int expect, int update)来进行操作，他们能保证状态的改变是安全的。子类被推荐定义为同步组件的静态内部类，同步器自身没有实现任何同步接口，它仅仅定义了若干同步状态获取和释放的方法来供自定义同步组件使用。同步器支持同步状态的独占获取或者共享获取，以方便实现不同类型的同步组件。

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

AQS的相关API：

　　同步器提供的可重写的方法：

方法名称	描述
boolean tryAcquire(int arg)	独占式获取同步状态。实现该方法需要查询当前状态并判断同步状态是否符合预期，然后再进行cas设置同步状态。（因为可能多个线程都在获取）
boolean tryRelease(int arg)	独占式释放同步状态，成功返回true，失败返回false。
int tryAcquireShared(int arg)	共享式获取同步状态，获取成功则返回值>=0
boolean tryReleaseShared(int arg)	共享式释放同步状态，成功返回true，失败返回false。
boolean isHeldExclusively()	判断同步器是否在独占模式下被占用，一般用来表示同步器是否被当前线程占用

　　同步器提供的操作状态的方法：

方法	描述
int getState()	获取当前同步状态
void setState(int newState)	设置当前同步状态
boolean compareAndSetState(int expect, int update)	使用CAS设置当前状态，保证状态更新的原子性。只有当state为expect时，才能设置成功。expect相当于乐观锁的version

　　同步器提供的模板方法：

方法	描述
void acquire(int arg)	独占式获取同步状态，该方法会调用子类重写的tryAcquire(int arg)，如果tryAcquire返回true则该方法直接返回，否则先将当前线程加入同步队列的尾部，然后阻塞当前线程
void acquireInterruptibly(int arg)	和acquire类似，只是当线程获取同步状态失败被阻塞后，可以响应中断，收到中断后将会取消获取同步状态
boolean tryAcquireNanos(int arg, long nanosTimeout)	在acquireInterruptibly的基础上加了超时限制，如果在超时时间内获取到同步状态返回true，否则返回false
boolean release(int arg)	独占式释放同步状态，该方法会在释放同步状态后将第一个节点（对应刚刚释放同步状态的线程）的后继节点对应的线程唤醒。
void acquireShared(int arg)	共享式获取同步状态，该方法会调用子类重写的tryAcquireShared(int arg)，如果tryAcquireShared返回true则该方法直接返回，否则先将当前线程加入同步队列的尾部，然后阻塞当前线程
void acquireSharedInterruptibly(int arg)	和acquireShared类似，只是当线程获取同步状态失败被阻塞后，可以响应中断，收到中断后将会取消获取同步状态
boolean tryAcquireSharedNanos(int arg, long nanosTimeout)	在acquireSharedInterruptibly的基础上加了超时限制，如果在超时时间内获取到同步状态返回true，否则返回false
boolean releaseShared(int arg)	共享式的释放同步状态
Collection<Thread> getQueuedThreads()	获取等待在同步队列上的线程集合

　　我们依据AQS来实现一个排它锁并进行测试：

排它锁代码：

public class Mutex implements Lock {

    private final Sync sync = new Sync();

    //静态内部类，自定义同步器，锁的功能通过它来实现

    private static class Sync extends AbstractQueuedSynchronizer{

        //是否处于占用状态

        protected boolean isHeldExclusively(){

            return getState() == 1;

        }

        //状态为0的时候，获取锁的方法

        public boolean tryAcquire(int acquires) {

            if(compareAndSetState(0,1)){

                setExclusiveOwnerThread(Thread.currentThread());

                return true;

            }

            return false;

        }

        //释放锁，将状态设置为0

        protected boolean tryRelease(int arg) {

            if(getState()==0){

                throw new IllegalMonitorStateException();

            }

            setExclusiveOwnerThread(null);//移除占用线程的引用

            setState(0);

            return true;

        }

        Condition newCondition(){return new ConditionObject();}

    }

    //---以下是锁的具体实现，都是通过sync实现的---------

    @Override

    public void lock() {

        sync.acquire(1);

    }

    @Override

    public void lockInterruptibly() throws InterruptedException {

        sync.acquireInterruptibly(1);

    }

    @Override

    public boolean tryLock() {

        return sync.tryAcquire(1);

    }

    @Override

    public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {

        return sync.tryAcquireNanos(1,unit.toNanos(timeout));

    }

    @Override

    public void unlock() {

        sync.release(1);

    }

    @Override

    public Condition newCondition() {

        return sync.newCondition();

    }

}

　　我们建多个线程，对int值count进行累加。测试代码：

public class ThreadTest {

    private int count;

    private Mutex mutex = new Mutex();

    @Test

    public void addTest1(){

        MyThread1 mt1 = new MyThread1("111");

        MyThread1 mt2 = new MyThread1("222");

        mt1.start();

        mt2.start();

        try {

            mt1.join();

            mt2.join();

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        System.out.println("addTest1  count = " + count);

    }

    @Test

    public void addTest2(){

        MyThread2 mt1 = new MyThread2();

        MyThread2 mt2 = new MyThread2();

        mt1.start();

        mt2.start();

        try {

            mt1.join();

            mt2.join();

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        System.out.println("addTest2  count = " + count);

    }

    class MyThread1 extends Thread{

        private String name;

        public MyThread1(String name){ this.name=name;}

        @Override

        public void run() {

            for(int i=0;i<1000;i++){

                count = count + 1;

            }

        }

    }

    class MyThread2 extends Thread{

        @Override

        public void run() {

            for(int i=0;i<1000;i++){

                mutex.lock();

                try {

                    count = count + 1;

                }catch (Exception e){

                    System.out.println("异常啦 ~ ~ " +e.getMessage());

                }finally {

                    mutex.unlock();

                }

            }

        }

    }

}

测试结果：

addTest1结果不确定，每次执行结果均小于2000，addTest2执行结果每次都是2000。显然，这是由于MyThread1中多线程同时执行，而累加过程又非原子操作，造成了内存覆盖导致。

巴特西

队列同步器AbstractQueuedSynchronizer

最新文章

热门文章