AQS

1、state

AbstractQueuedSynchronizer维护了一个volatile int类型的变量，用户表示当前同步状态。

private volatile int state;

    protected final int getState() {
        return state;
    }

    protected final void setState(int newState) {
        state = newState;
    }

    protected final boolean compareAndSetState(int expect, int update) {
        // See below for intrinsics setup to support this
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

这三个方法都是原子性操做。

其中AQS类继承AbstractOwnableSynchronizer

private transient Thread exclusiveOwnerThread;

    protected final void setExclusiveOwnerThread(Thread thread) {
        exclusiveOwnerThread = thread;
    }

    protected final Thread getExclusiveOwnerThread() {
        return exclusiveOwnerThread;
    }

当中的这两个方法能够分别记录当前AQS所同步着的线程。

2、acquire(int)——以独占方式加锁

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

tryAcquire()尝试直接去获取资源，若是成功则直接返回；(tryAcquire是抽象方法，暴露给子类实现)

addWaiter()方法负责把当前没法得到锁的线程包装为一个Node添加到队尾，SHARED：共享锁，EXCLUSIVE：独占锁。

static final Node SHARED = new Node();

    static final Node EXCLUSIVE = null;    

    private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

acquireQueued()的主要做用是把已经追加到队列的线程节点经过调用parkAndCheckInterrupt()进行阻塞，等被唤醒后再去竞争资源。

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

检查前一个节点的线程状态

• 若是前继的节点状态为SIGNAL，代表当前节点须要unpark，则返回成功，此时parkAndCheckInterrupt将致使线程阻塞
• 若是前继节点状态为CANCELLED(ws>0)，说明前置节点已经被放弃，则回溯到一个非取消的前继节点，返回false，acquireQueued方法的无限循环将递归调用该方法，直至规则1返回true，致使线程阻塞
• 若是前继节点状态为非SIGNAL、非CANCELLED，则设置前继的状态为SIGNAL，返回false后进入acquireQueued的无限循环，与规则2同
• 整体看来，shouldParkAfterFailedAcquire就是靠前继节点判断当前线程是否应该被阻塞，若是前继节点处于CANCELLED状态，则顺便删除这些节点从新构造队列。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        if (ws > 0) {
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

3、release(int)——以独占方式解锁

release：若是能够释放锁，则唤醒队列第一个线程（Head）

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

tryRelease：若是线程屡次锁定，则进行屡次释放，直至status==0则真正释放锁，所谓释放锁即设置status为0，由于无竞争因此没有使用CAS。 

protected final boolean tryRelease(int releases) {
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }

private void unparkSuccessor(Node node) {
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        /*
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }

4、ReentrantLock()的实现

ReentrantLock()有公平锁和非公平锁两种

public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

    public void lock() {
        sync.lock();
    }

非公平锁： 

static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        /**
         * Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.
         */
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }

    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }

• 该方法会首先判断当前状态，若是c==0说明没有线程正在竞争该锁，若是不c !=0 说明有线程正拥有了该锁
• 若是发现c==0，则经过CAS设置该状态值为acquires,acquires的初始调用值为1
• 若是c !=0 但发现本身已经拥有锁，只是简单地++acquires，并修改status值，这就是可重入性的原理

 

公平锁：

static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;

        final void lock() {
            acquire(1);
        }

        /**
         * Fair version of tryAcquire.  Don't grant access unless
         * recursive call or no waiters or is first.
         */
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

hasQueuedPredecessors()就是判断锁是否公平的关键，若是在当前线程以前还有排队的线程就返回true，这时候当前线程就不会去竞争锁。从而保证了锁的公平性。

5、读写锁

读锁的获取与释放

• tryAcquireShared()尝试获取资源，成功则直接返回；
• 失败则经过doAcquireShared()进入等待队列park()，直到被unpark()/interrupt()并成功获取到资源才返回。整个等待过程也是忽略中断的。

public void lock() {
        sync.acquireShared(1);
    }

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

    protected final int tryAcquireShared(int unused) {

        Thread current = Thread.currentThread();
        int c = getState();

        //获取写锁，写锁存在且不是当前线程
        if (exclusiveCount(c) != 0 &&
            getExclusiveOwnerThread() != current)
            return -1;
        //获取读锁
        int r = sharedCount(c);
        
        if (!readerShouldBlock() &&
            r < MAX_COUNT && //MAX_COUNT为获取读锁的最大数量，为16位的最大值
            compareAndSetState(c, c + SHARED_UNIT)) {
            if (r == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                HoldCounter rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    cachedHoldCounter = rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;
            }
            return 1;
        }
        return fullTryAcquireShared(current);
    }


    private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        //若是有资源，继续释放后面的节点
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

tryAcquireShared()流程：

1. 经过同步状态低16位判断，若是存在写锁且当前线程不是获取写锁的线程，返回-1，获取读锁失败；不然执行步骤2）。
2. 经过readerShouldBlock判断当前线程是否应该被阻塞，若是不该该阻塞则尝试CAS同步状态；不然执行3）。
3. 第一次获取读锁失败，经过fullTryAcquireShared再次尝试获取读锁。

doAcquireShared()：

tryAcquireShared()尝试加锁失败，调用doAcquireShared()加入阻塞队列以前，还会尝试一次加锁，有剩余的话还会唤醒以后的队友。那么问题就来了，假如老大用完后释放了5个资源，而老二须要6个，老三须要1个，老四须要2个。老大先唤醒老二，老二一看资源不够，他是把资源让给老三呢，仍是不让？答案是否认的！老二会继续park()等待其余线程释放资源，也更不会去唤醒老三和老四了。独占模式，同一时刻只有一个线程去执行，这样作何尝不可；但共享模式下，多个线程是能够同时执行的，如今由于老二的资源需求量大，而把后面量小的老三和老四也都卡住了。固然，这并非问题，只是AQS保证严格按照入队顺序唤醒罢了（保证公平，但下降了并发）