java并发之AQS

一、简介

AQS，全称：AbstractQueuedSynchronizer，是一个JDK提供的用于构建锁、同步器等线程协作工具类的框架，内部维护FIFO双向队列（双向链表实现）。
AQS重要属性：

// 表示同步状态。它既可以表示独占模式下的锁状态，也可以表示共享模式下的资源数量。通过修改state字段，可以实现多线程的独占或共享模式‌
private volatile int state
// 当前持有独占锁的线程
private transient Thread exclusiveOwnerThread
// 头节点
private transient volatile Node head;
// 尾节点
private transient volatile Node tail;

Node节点重要属性：

// 加入队列的线程
volatile Thread thread;
// 前驱节点
volatile Node prev;
// 后继节点
volatile Node next;
// CANCELLED: 表示线程已经取消了对同步状态的请求。
// SIGNAL: 表示线程需要被唤醒（通常是因为其他线程释放了同步状态）。
// CONDITION: 表示线程正在等待某个条件。
// PROPAGATE: 表示下一次共享状态的释放应该传播到其他线程。
// 0: 初始状态，表示节点没有特定的状态。
volatile int waitStatus;
Node nextWaiter;

AQS 在 ReentrantLock、ReentrantReadWriteLock、Semaphore、CountDownLatch、ThreadPoolExcutor 的 Worker 中都有运用（JDK 1.8），AQS 是这些类的底层原理。

二、实现自定义线程协作工具类

2.1 实现独占锁

重写AQS以下方法

boolean tryAcquire(int arg)
boolean tryRelease(int arg)
boolean isHeldExclusively()

调用AQS以下方法

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

2.2 实现共享锁

重写AQS以下方法

int tryAcquireShared(int arg)
boolean tryReleaseShared(int arg)

调用AQS以下方法

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

2.3 示例

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class Test {

    class MySync extends AbstractQueuedSynchronizer {

        @Override
        protected boolean tryAcquire(int arg) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

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

    class MyLock implements Lock {

        private MySync sync = new MySync();

        @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 time, TimeUnit unit) throws InterruptedException {
            return sync.tryAcquireNanos(1, unit.toNanos(time));
        }

        @Override
        public void unlock() {
            sync.release(1);
        }

        @Override
        public Condition newCondition() {
            return sync.newCondition();
        }
    }

    public static void main(String[] args) throws Exception {
        Test test = new Test();
        MyLock myLock = test.new MyLock();

        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    myLock.lock();
                    System.out.println(Thread.currentThread().getName() + "执行开始");
                    Thread.sleep(5000L);
                    System.out.println(Thread.currentThread().getName() + "执行结束");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    myLock.unlock();
                }
            }
        }, "t1");

        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    myLock.lock();
                    System.out.println(Thread.currentThread().getName() + "执行开始");
                    Thread.sleep(3000L);
                    System.out.println(Thread.currentThread().getName() + "执行结束");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    myLock.unlock();
                }
            }
        }, "t2");

        Thread t3 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    myLock.lock();
                    System.out.println(Thread.currentThread().getName() + "执行开始");
                    Thread.sleep(1000L);
                    System.out.println(Thread.currentThread().getName() + "执行结束");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    myLock.unlock();
                }
            }
        }, "t3");

        t1.start();
        t2.start();
        t3.start();
    }
}

• 参考1