JUC同步锁原理源码解析一 之ReentrantLock

JUC同步锁原理

1.锁的本质

1.什么是锁？

? 通俗来讲，锁要保证的就是原子性，就是一个代码块不允许多线程同时执行,就是锁。从生活的角度上来说，就比如你要去上厕所，当你在上厕所期间，你会把门锁上，其他人只能排队。不允许多个人同时上厕所。

2.锁的底层实现

? java语言是运行在jvm之上，jvm是由C++实现的。java本身没有对应的底层锁实现，它将锁的问题抛给了C++。C++将锁的实现抛给了汇编语言，汇编语言将问题抛给操作系统。所以最后还是由操作系统提供的cmpxchg指令。通过 lock cmpxchg指令实现了cpu对于单个变量的原子操作。lock cmpxchg涉及到缓存一致性协议(MESI)与总线锁。（个人能力有限，自行了解吧）

3.什么是自旋锁

? 线程不停执行某一个代码块，直到满足条件或者操作重试次数，也就是我们所说的CAS。

4.如何实现一把锁

1.状态：判断当前是有锁还是无锁？

? boolean state = true/false;标识有锁与无锁。但是一旦有锁冲入的情况，我们就需要引入一个新的变量去存储锁冲入的次数。所以JUC中使用int state,默认值为0代表无锁状态。上增的数值代表所冲入的次数。

2.多线程如何抢锁

? 通过cas实现多线程抢锁

3.抢不到锁的线程如何处理

? 1.自旋，继续抢锁直到抢成功

? 2.阻塞，线程阻塞直到有线程唤醒

? 3.自旋+阻塞，自旋抢锁一定次数，如果失败，线程阻塞。

4.自旋锁的优缺点

? 1.优点：节省线程上线文切换的时间。适用于执行步骤少且快的场景，节省cpu资源

? 2.缺点：占用cpu资源，消耗cpu性能

? 注意点：当cpu个数增加且线程数增加，可能导致自旋锁的优点退化成缺点。

2.AQS源码

Node节点

 static final class Node {
        /** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
 
        static final int PROPAGATE = -3;

        volatile int waitStatus;

        volatile Node prev;

        volatile Node next;
       
        volatile Thread thread;

        Node nextWaiter;
}

AbstractQueuedSynchronizer类

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
 	private transient volatile Node head;

    /**
     * Tail of the wait queue, lazily initialized.  Modified only via
     * method enq to add new wait node.
     */
    private transient volatile Node tail;

    /**
     * The synchronization state.
     */
    private volatile int state;//最重要的一个变量
       
}

ConditionObject类

public class ConditionObject implements Condition, java.io.Serializable {
        private static final long serialVersionUID = 1173984872572414699L;
        /** First node of condition queue. */
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        private transient Node lastWaiter;
}

accquire方法

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&//尝试获取锁
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//如果获取锁失败，添加到队列中，由于ReentrantLock是独占锁所以节点必须是EXCLUSIVE类型
        selfInterrupt();//添加中断标识位
}

addWaiter方法

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);//cas失败后执行入队操作，继续尝试
     return node;
 }

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;//获取尾指针
        if (t == null) { //代表当前队列没有节点
            if (compareAndSetHead(new Node()))//将当前节点置为头结点
                tail = head;
        } else {//当前队列有节点
            node.prev = t;//
            if (compareAndSetTail(t, node)) {//将当前节点置为尾结点
                t.next = node;
                return t;
            }
        }
    }
}

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)) {//前驱节点等于头节点尝试cas抢锁。
                setHead(node);//抢锁成功将当前节点设置为头节点
                p.next = null; // help GC  当头结点置空
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&//当队列中有节点在等待，判断是否应该阻塞
                parkAndCheckInterrupt())//阻塞等待，检查中断标识位
                interrupted = true;//将中断标识位置为true
        }
    } finally {
        if (failed)//
            cancelAcquire(node);//取消当前节点
    }
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;//获取上一个节点的等待状态
    if (ws == Node.SIGNAL)//如果状态为SIGNAL,代表后续节点有节点可以唤醒，可以安心阻塞去
        /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
        return true;
    if (ws > 0) {//如果当前状态大于0，代表节点为CANCELLED状态
        /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
        do {
            node.prev = pred = pred.prev;//从尾节点开始遍历，找到下一个状态不是CANCELLED的节点。将取消节点断链移除
        } 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.
             */
        //这里需要注意ws>0时，已经找到了一个不是取消状态的前驱节点。
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);//将找到的不是CANCELLED节点的前驱节点，将其等待状态置为SIGNAL
    }
    return false;
}

 private void cancelAcquire(Node node) {
     // Ignore if node doesn't exist
     if (node == null)//当前节点为空直接返回
         return;

     node.thread = null;//要取消了将当前节点的线程置为空
     // Skip cancelled predecessors
     Node pred = node.prev;//获取到当前节点的前驱节点
     while (pred.waitStatus > 0)//如果当前节点的前驱节点的状态大于0,代表是取消状态，一直找到不是取消状态的节点
         node.prev = pred = pred.prev;
     Node predNext = pred.next;//将当前要取消的节点断链

     node.waitStatus = Node.CANCELLED;//将当前节点的等待状态置为CANCELLED
     // If we are the tail, remove ourselves.
     if (node == tail && compareAndSetTail(node, pred)) {//如果当前节点是尾结点，将尾结点替换为浅语节点
         compareAndSetNext(pred, predNext, null);//将当前节点的下一个节点置为空，因为当前节点是最后一个节点没有next指针
     } else {
         // If successor needs signal, try to set pred's next-link
         // so it will get one. Otherwise wake it up to propagate.
         int ws;
         if (pred != head &&//前驱节点不等于头结点
             ((ws = pred.waitStatus) == Node.SIGNAL ||//前驱节点的状态不等于SIGNAL
              (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驱节点的状态小于0，并且cas将前驱节点的等待置为SIGNAL
             pred.thread != null) {//前驱节点的线程补位空
             Node next = node.next;//获取当前节点的next指针
             if (next != null && next.waitStatus <= 0)//如果next指针不等于空并且等待状态小于等于0，标识节点有效
                 compareAndSetNext(pred, predNext, next);//将前驱节点的next指针指向下一个有效节点
         } else {
             unparkSuccessor(node);//唤醒后续节点 条件：1.前驱节点是头结点 2.当前节点不是signal,在ReentransLock中基本不会出现，在读写锁时就会出现
         }

         node.next = node; // help GC 将引用指向自身
     }
 }

 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);//cas将当前节点置为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;//将s置为空
         for (Node t = tail; t != null && t != node; t = t.prev)//从尾结点遍历找到一个不是取消状态的节点
             if (t.waitStatus <= 0)
                 s = t;
     }
     if (s != null)//如果s不等于空
         LockSupport.unpark(s.thread);//唤醒当前节点s
 }

总结：AQS提供了统一的模板，对于如何入队出队以及线程的唤醒都由AQS提供默认的实现，只需要子类实现自己上锁和解锁的逻辑。

3.ReentrantLock

基本使用

public class ReentrantLock extends Thread {

	private static ReentrantLock lock=new ReentrantLock(true); //参数为true表示为公平锁，请对比输出结果
    public void run() {
        for(int i=0; i<100; i++) {
            lock.lock();
            try{
                System.out.println(Thread.currentThread().getName()+"获得锁");
            }finally{
                lock.unlock();
            }
        }
    }
    public static void main(String[] args) {
        ReentrantLock rl=new ReentrantLock();
        Thread th1=new Thread(rl);
        Thread th2=new Thread(rl);
        th1.start();
        th2.start();
    }
}

lock方法:

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

在AQS中对于具体的lock方法，并不做具体的实现，由子类自由拓展。ReentrantLock中lock方法中的实现分为公平锁和非公平锁的实现。所以lock方法有两个实现。

公平锁FairSync的实现

final void lock() {
    acquire(1);//获取锁
}

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&//尝试获取锁
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//如果获取锁失败，添加到队列中，由于ReentrantLock是独占锁所以节点必须是EXCLUSIVE类型
        selfInterrupt();//添加中断标识位
}

tryAcquire方法

protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();//获取到当前线程
    int c = getState();//获取当前的同步状态值
    if (c == 0) {//代表没有线程占有锁
        if (!hasQueuedPredecessors() &&//是否有前驱节点
            compareAndSetState(0, acquires)) {//如果没有前驱节点，cas将当前状态值置为acquires,也就是1，成功代表获取到锁
            setExclusiveOwnerThread(current);//标识当前属于互斥状态线程的拥有者是当前线程
            return true;//true，代表获取锁成功
        }
    }
    else if (current == getExclusiveOwnerThread()) {//进入这里代表state不为0,有其他线程获得锁
        int nextc = c + acquires;//锁冲入，将冲入次数加1
        if (nextc < 0)//冲入次数不能少于0，少于0是非法值，抛出异常
            throw new Error("Maximum lock count exceeded");
        setState(nextc);//设置state状态值
        return true;//true，代表获取锁成功
    }
    return false;//返回false，代表获取锁失败
}

 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; // 获取队列的尾指针
     Node h = head;// 获取队列的头指针
     Node s;
     return h != t &&//如果头结点和尾结点不是同一个节点
         ((s = h.next) == null || s.thread != Thread.currentThread());//头结点的下一个节点为空或者当前头结点的线程不等于当前线程。
 }

addWaiter方法

 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);//cas失败后执行入队操作，继续尝试
     return node;
 }

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;//获取尾指针
        if (t == null) { //代表当前队列没有节点
            if (compareAndSetHead(new Node()))//将当前节点置为头结点
                tail = head;
        } else {//当前队列有节点
            node.prev = t;//
            if (compareAndSetTail(t, node)) {//将当前节点置为尾结点
                t.next = node;
                return t;
            }
        }
    }
}

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)) {//前驱节点等于头节点尝试cas抢锁。
                setHead(node);//抢锁成功将当前节点设置为头节点
                p.next = null; // help GC  当头结点置空
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&//当队列中有节点在等待，判断是否应该阻塞
                parkAndCheckInterrupt())//阻塞等待，检查中断标识位
                interrupted = true;//将中断标识位置为true
        }
    } finally {
        if (failed)//
            cancelAcquire(node);//取消当前节点
    }
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;//获取上一个节点的等待状态
    if (ws == Node.SIGNAL)//如果状态为SIGNAL,代表后续节点有节点可以唤醒，可以安心阻塞去
        /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
        return true;
    if (ws > 0) {//如果当前状态大于0，代表节点为CANCELLED状态
        /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
        do {
            node.prev = pred = pred.prev;//从尾节点开始遍历，找到下一个状态不是CANCELLED的节点。将取消节点断链移除
        } 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.
             */
        //这里需要注意ws>0时，已经找到了一个不是取消状态的前驱节点。
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);//将找到的不是CANCELLED节点的前驱节点，将其等待状态置为SIGNAL
    }
    return false;
}

 private void cancelAcquire(Node node) {
     // Ignore if node doesn't exist
     if (node == null)//当前节点为空直接返回
         return;

     node.thread = null;//要取消了将当前节点的线程置为空
     // Skip cancelled predecessors
     Node pred = node.prev;//获取到当前节点的前驱节点
     while (pred.waitStatus > 0)//如果当前节点的前驱节点的状态大于0,代表是取消状态，一直找到不是取消状态的节点
         node.prev = pred = pred.prev;
     Node predNext = pred.next;//将当前要取消的节点断链

     node.waitStatus = Node.CANCELLED;//将当前节点的等待状态置为CANCELLED
     // If we are the tail, remove ourselves.
     if (node == tail && compareAndSetTail(node, pred)) {//如果当前节点是尾结点，将尾结点替换为浅语节点
         compareAndSetNext(pred, predNext, null);//将当前节点的下一个节点置为空，因为当前节点是最后一个节点没有next指针
     } else {
         // If successor needs signal, try to set pred's next-link
         // so it will get one. Otherwise wake it up to propagate.
         int ws;
         if (pred != head &&//前驱节点不等于头结点
             ((ws = pred.waitStatus) == Node.SIGNAL ||//前驱节点的状态不等于SIGNAL
              (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驱节点的状态小于0，并且cas将前驱节点的等待置为SIGNAL
             pred.thread != null) {//前驱节点的线程补位空
             Node next = node.next;//获取当前节点的next指针
             if (next != null && next.waitStatus <= 0)//如果next指针不等于空并且等待状态小于等于0，标识节点有效
                 compareAndSetNext(pred, predNext, next);//将前驱节点的next指针指向下一个有效节点
         } else {
             unparkSuccessor(node);//唤醒后续节点 条件：1.前驱节点是头结点 2.当前节点不是signal,在ReentransLock中基本不会出现，在读写锁时就会出现
         }

         node.next = node; // help GC 将引用指向自身
     }
 }

 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);//cas将当前节点置为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;//将s置为空
         for (Node t = tail; t != null && t != node; t = t.prev)//从尾结点遍历找到一个不是取消状态的节点
             if (t.waitStatus <= 0)
                 s = t;
     }
     if (s != null)//如果s不等于空
         LockSupport.unpark(s.thread);//唤醒当前节点s
 }

非公平锁FairSync的实现

lock方法

final void lock() {
    if (compareAndSetState(0, 1))//上来直接抢锁。
        setExclusiveOwnerThread(Thread.currentThread());//抢锁成功，将当前互斥锁的拥有线程设置为当前线程
    else
        acquire(1);//cas失败去acquire
}

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&//尝试获取锁
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//如果获取锁失败，添加到队列中，由于ReentrantLock是独占锁所以节点必须是EXCLUSIVE类型
        selfInterrupt();//添加中断标识位
}

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

nonfairTryAcquire方法

 final boolean nonfairTryAcquire(int acquires) {
     final Thread current = Thread.currentThread();//获取到当前线程
     int c = getState();//获取当前状态
     if (c == 0) {//代表没有锁
         if (compareAndSetState(0, acquires)) {//cas自旋尝试获得锁
             setExclusiveOwnerThread(current);//当前独占锁的拥有线程设置为当前线程
             return true;//返回
         }
     }
     else if (current == getExclusiveOwnerThread()) {//如果当前线程之前已经获取到锁
         int nextc = c + acquires;//锁重入
         if (nextc < 0) // overflow 锁重入不能为0
             throw new Error("Maximum lock count exceeded");
         setState(nextc);//设置最新的state状态
         return true;
     }
     return false;
 }

acquireQueued方法和addWaiter方法与公平锁的实现一致

锁释放

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

 public final boolean release(int arg) {
     if (tryRelease(arg)) {//尝试释放锁
         Node h = head;
         if (h != null && h.waitStatus != 0)//如果头结点不为空并且头结点的等待状态不等于0
             unparkSuccessor(h);//唤醒头结点
         return true;
     }
     return false;
 }

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;//当前的状态减去释放锁的数量
    if (Thread.currentThread() != getExclusiveOwnerThread())//如果当前线程不是独占锁的线程，没锁还要释放，不就抛出异常了吗
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {//状态为0
        free = true;
        setExclusiveOwnerThread(null);//将当前互斥锁的拥有线程设置为空
    }
    setState(c);//设置状态位
    return free;
}

4.留言

本文章只是JUC 中AQS的一部分，后续的文章会对基于AQS锁实现的子类进行拓展讲解，以上文章内容基于个人以及结合别人文章的理解，如果有问题或者不当之处欢迎大家留言交流。由于为了保证观看流畅性，其中一部分源码有重复的地方。请见谅