JUC并发编程--------AQS以及各类锁

AQS 定义了5个队列中节点状态：

ReentrantLock是一种可重入的独占锁，它允许同一个线程多次获取同一个锁而不会被阻塞。

它的功能类似于synchronized是一种互斥锁，可以保证线程安全。相对于 synchronized，

ReentrantLock具备如下特点：

• 可中断
• 可以设置超时时间
• 可以设置为公平锁
• 支持多个条件变量
• 与 synchronized 一样，都支持可重入

它的主要应用场景是 在多线程环境下对共享资源进行独占式访问，以保证数据的一致性和安全性。

在使用时要注意 4 个问题：

ReentrantLock支持公平锁和非公平锁两种模式：

公平锁：线程在获取锁时，按照等待的先后顺序获取锁。

非公平锁：线程在获取锁时，不按照等待的先后顺序获取锁，而是随机获取锁。ReentrantLock默认是非公平锁

1 ReentrantLock lock = new ReentrantLock(); //参数默认false，不公平锁
2 ReentrantLock lock = new ReentrantLock(true); //公平锁
 
 
//非公平锁
     final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
 
//公平锁
final void lock() {
            acquire(1);
        }
 
//两者的区别在于非公平锁刚进来则会先用当前线程去尝试获取锁，如果获取失败再走acquire流程
 
 
 

可重入锁

 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;
        }

源码解析

公平锁：

首先先看公平锁类FairSync，其中包含了lock（） 以及tryAcquire（int acquires） 方法

    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;
        }
    }

要加锁的时候我们首先调用lock方法， 其中会调用acquire（1）方法

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

调用acquire方法之后，由于我们使用的是公平锁，紧接着就会先调用公平锁的tryAcquire方法

 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;
        }
    }

进入tryAcquire方法之后

首先先通过getState获取状态，当状态值为0的时候，表示锁未被持有，于是进入

if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current)

的判断，在hasQueuedPredecessors()方法中，会判断线程是否有等待阻塞队列，如果有等待阻塞队列，并且头节点是当前线程，则返回false，加上'!'的非判断，就可以进入 compareAndSetState(0, acquires)，尝试修改锁的状态为持有，如果成功则调用 setExclusiveOwnerThread(current);方法将当前线程设为锁持有线程。

如果当前状态不为0的时候，表示锁已经被持有，这时候通过if (current == getExclusiveOwnerThread())方法判断锁的持有则是不是当前线程，如果是当前线程，则将锁的持有状态+1，代表重入锁

  public final boolean hasQueuedPredecessors() {
        // The correctness of this depends on head being initialized
        // before tail and on head.next being accurate if the current
        // thread is first in queue.
        Node t = tail; // Read fields in reverse initialization order
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }

if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current)

只有该方法返回false的情况，才能进入下一步通过cas获取锁。

在该方法中：

判断h！=t，表示头节点不等于尾节点，才能继续判断，否则就是头节点等于尾节点表示阻塞队列为空或则阻塞队列只有一个线程，那么可以直接返回false

当h！=t的时候，表示存在阻塞队列，并且存在多个线程在等待，此时需要继续判断，当s=h.next）==null的时候，表示还不存在阻塞队列，则返回true，需要先创建阻塞队列

当s=h.next）！=null的时候，表示已经存在阻塞队列了，这时候需要进行下一步判断

当s.thread==Thread.currentThread()的时候，表示头节点的下一个节点是当前线程，则返回false准备尝试获取锁

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

掉调用tryAcquire方法之后，如果没有成功获取锁，就会将进入 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))方法

进入acquireQueued方法之前，先通过addWaiter(Node.EXCLUSIVE)用当前线程入参创建一个节点。

 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;
    }

通过 Node node = new Node(Thread.currentThread(), mode);将当前线程放入新创的节点中，从上一个方法可以看出传入的mode是Node.EXCLUSIVE，表示独占锁

创建好节点之后，首先判断tail尾节点是否为为空，如果不为空，则将当前节点的前驱节点设置为尾节点，然后通过compareAndSetTail(pred, node)方法，尝试将aqs维护的尾节点换成刚创建的界定啊，如果cas成功，则将原先的尾节点的后序节点设置成当前节点，至此新创建的节点变为尾节点。

如果tail==null，表示还没存在尾节点，这时候需要调用enq(node);方法

   private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

进入enq方法之后，会循环创建尾节点，继续通过tail==null进行二次判断，如果当前尾节点还是null，则cas新建一个节点作为头节点（初始化），完成之后再设置尾节点等于头节点。

然后进入第二次循环，此时tail肯定！=null，则将传入进来的新节点的前驱节点设置为尾节点，然后尝试交换尾节点，成功之后继续将原先的尾节点的后序节点设置为传入的新节点，至此，新线程入队成功，此时addWaiter方法调用完毕

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

执行完addwaiter方法之后，接着执行acquireQueued方法

   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);
        }
    }

在该方法中，通过 final Node p = node.predecessor();获取当前线程节点的前驱节点，

如果前驱节点是头节点，表示当前节点排第二，可以通过tryAcquire获取锁，获取锁成功，则通过setHead(node);方法，将节点线程设置为null（因为之后不在需要阻塞与唤醒了），

并且返回fase（该返回值有特别的意义，如果返回true，方法结束之后会接着调用selfInterrupt();方法，将中断标记位设置为true，后面不会在被park中断），如果获取锁失败，就会进入shouldParkAfterFailedAcquire(p, node)方法判断是否需要进行中断，如果需要中断则调用parkAndCheckInterrupt()方法中断，中断结束之后就会将interrupted设置为true并返回。

  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;
    }

该方法是为了判断是否需要进行中断，比较巧妙，

首先进来会先判断前驱节点的状态，默认都是0，而该方法的目的就是要即将前驱节点状态设置成-1，在-1状态下，如果前驱节点释放锁，就会激活该节点，然后重新尝试获取锁。

从源码上可以看到，当线程调用lock的时候，进来这个方法，前驱节点都是0，然后设置完前驱节点的状态为-1之后，会返回一个false，在外层又会进行一次自旋，如果前驱节点是头节点，则可以尝试通过cas尝试获取锁，当获取锁失败再次进来的时候，通过 if (ws == Node.SIGNAL)判断成功，直接返回ture，表示需要中断

    /**
     * Convenience method to park and then check if interrupted
     *
     * @return {@code true} if interrupted
     */
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

线程park然后被中断之后，执行Thread.interrupted();清空中断标记位，避免下次无法中断造成不停的自旋不断消耗cpu，至此，lock全过程就完成了

获取锁完成业务之后开始释放锁

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

调用unlock方法，然后调用release方法，传入状态值为1

   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;
        }

先通过tryRelease方法将state-1（加锁的时候是+1，释放锁则需要-1还原），如果状态值为0，则表示释放锁成功，此时通过   setExclusiveOwnerThread(null);将锁持有线程设置为null

如果不为0，则及继续返回一个false，表示锁还没有释放完成，仍然被持有（重入锁状态）；

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

如果释放锁成功，返回ture之后，进入方法体，判断头节点是否为为空，此时头节点就是自己，肯定不为空。并且waitStatus在下一个线程进入阻塞队列之前就已经被设置为-1，所以顺利执行unparkSuccessor(h);方法

    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);
    }

进入该方法后，waitStatus的值为-1，小于0，于是开始通过cas将waitStatus设置为0。

紧接着尝试找到当前节点的下一个节点，如果不为null，则通过unpark唤醒该线程