ASQ分析1-基于AQS自定义锁分析

秒速五厘米 2022-12-02 00:46 173阅读 0赞

## 自定义锁 ##

`AQS`是并发包非常重要的基础锁类,我们看下如何利用其快速实现一个`锁`.

public class SimpleLock extends AbstractQueuedSynchronizer {
        
        @Override
        protected boolean tryAcquire(int arg) {
        
            boolean compareAndSetState = compareAndSetState(0, 1);
            if (compareAndSetState) {
        
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
        @Override
        protected boolean tryRelease(int arg) {
        
            if (getState() == 0) {
        
                throw new IllegalStateException();
            }
            if (getExclusiveOwnerThread() != Thread.currentThread()) {
        
                throw new IllegalStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }
        public void lock() {
        
            acquire(1);
        }
        public void unLock() {
        
            release(1);
        }
    }

上面的利用的AQS实现的一个，运行示例如下：

public class Main {
        
        public static void main(String[] args) {
        
            SimpleLock lock = new SimpleLock();
            //粗略的让子线程执行完成
            int i = 0;
            while (++i < 2000000) {
        
                Thread thread = new Thread(() ->
                {
        
                    lock.lock();
                    System.out.println(Thread.currentThread() + " 成功上锁");
                    try {
        
                        TimeUnit.MILLISECONDS.sleep(20);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                    lock.unLock();
                    System.out.println(Thread.currentThread() + " 成功解锁");
                }, "线程[" + i + "]");
                thread.start();
            }
            try {
        
                TimeUnit.HOURS.sleep(1);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }

输出：

Thread[线程[1],5,main] 成功上锁
    Thread[线程[354],5,main] 成功上锁
    Thread[线程[1],5,main] 成功解锁
    Thread[线程[354],5,main] 成功解锁
    Thread[线程[799],5,main] 成功上锁
    Thread[线程[799],5,main] 成功解锁
    Thread[线程[2],5,main] 成功上锁
    Thread[线程[2],5,main] 成功解锁
    Thread[线程[3],5,main] 成功上锁
    Thread[线程[3],5,main] 成功解锁
    Thread[线程[2195],5,main] 成功上锁
    Thread[线程[2195],5,main] 成功解锁

可见实现的锁为非公平锁。

### 加锁流程 ###

从`lock`开始分析：

//SimpleLock
     public void lock() {
        
            acquire(1);
     }

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

`tryAcquire`为继承者实现函数，继承需要实现`tryAcquire`判断当前线程是否能获取到锁，如果能获取返回true。否则调用`addWaiter(Node.EXCLUSIVE), arg)`将其将入`CLH`队列中,`acquireQueued`自旋锁不断重试调用`tryAcquire`获得锁。

大致流程如下图（不考虑中断等细节）：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center]

**`tryAcquire`说明：**

> 自定义锁需要实现的抽象函数，判断当前线程是否能获取到锁。

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

上诉操作涉及到AQS提供的一些函数和方法。所以简单介绍下AQS包含的属性和内置函数。  
我们首先看一个比较完整的类结构图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 1]  
由于上图非常冗余所以只提出部分属性和方法来做讲解：  
本例中用到属性和方法如下所示：

class AbstractQueuedSynchronizer {
        
        /**
         * 持有独占锁的线程
         */
        private transient Thread exclusiveOwnerThread;
    
        protected final void setExclusiveOwnerThread(Thread thread) {
        
            exclusiveOwnerThread = thread;
        }
    
        protected final Thread getExclusiveOwnerThread() {
        
            return exclusiveOwnerThread;
        }
    
        /**
         * cas方式设置状态
         */
        protected final boolean compareAndSetState(int expect, int update) {
        
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
    
        /**
         * 可以表示加锁的状态，比如1可以表示有一个线程获取到了锁，
         * 0表示没有线程获取到锁，这个解释意义是自类自行定义的。
         * 你也可以0表示加锁，1表示无人请求锁都行。
         */
        private volatile int state;
    	
    	/**
         * CLH头节点
         */
        private transient volatile Node head;
    
        /**
         * 
         * CLH尾节点
         */
        private transient volatile Node tail;
    }

寓意比较简单便不再过多解释，在回头看看我们自定义的`tryAcquire`函数就简单多了，首先明确  
这个函数的用途，如果能获取到锁函数返回`true` 否则返回`false`

//SimpleLock
      @Override
        protected boolean tryAcquire(int arg) {
        
        	//使用cas设置当前状态为1
            boolean compareAndSetState = compareAndSetState(0, 1);
            //设置成功
            if (compareAndSetState) {
        
                //设置独占锁的属性为当前线程 并返回true
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            //cas操作失败，证明已经有线程持有锁
            return false;
        }

假设`tryAcquire`返回false我们继续看`addWaiter`函数。  
`addWaiter`函数用于当线程无法获取锁时，让当前线程构造成一个`Node`类然后放入CLH队列中，关于CLH队列可以参看博主其他文章。

我们首先看下`Node`类的结构图:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 2]  
看着很复杂没关系我们只讲我们本小结用到的。

final class Node {
        
    	
    	//用于赋值给本类的nextWaiter属性
    	//表示当前节点所在锁是共享锁，还是排它锁
    	//SHARED表示当前锁是共享锁，多个线程可以同时获取到锁
        static final Node SHARED = new Node();
        //用于赋值给本类的nextWaiter属性
    	//表示当前节点所在锁是共享锁，还是排它锁
        //EXCLUSIVE表示当前节点所在锁为排它锁
        static final Node EXCLUSIVE = null;
        
        //waitStatus设置属性值。
        //当前节点因为各种原因被取消，比如中断等
        static final int CANCELLED = 1;
        //当前节点的子节点被阻塞，需要当前节点唤醒
        static final int SIGNAL = -1;
        //略
        static final int CONDITION = -2;
        //略
        static final int PROPAGATE = -3;
    
    
        //当前节点等候方式，遇到在细说
        //枚举常量为本类内部常量 CANCELLED SIGNAL PROPAGATE CONDITION
        //大于0表示被取消
        //当前节点唤醒或者解锁后 根据waitStatus来进行下一步的操作？？？
        volatile int waitStatus;
    
        //上一个节点（父节点）
        volatile Node prev;
        //下一节点(子节点)
        volatile Node next;
        //当前节点对应的线程
        volatile Thread thread;
    
        //下一个节点的等候方式，遇到在细说
        Node nextWaiter;
    
        final boolean isShared() {
        
            return nextWaiter == SHARED;
        }
    
    
        final Node predecessor() throws NullPointerException {
        
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
        }
    
        Node() {
            // Used to establish initial head or SHARED marker
        }
    
        Node(Thread thread, Node mode) {
             // Used by addWaiter
            this.nextWaiter = mode;
            this.thread = thread;
        }
    
        Node(Thread thread, int waitStatus) {
         // Used by Condition
            this.waitStatus = waitStatus;
            this.thread = thread;
        }
    }

回头继续看加锁流程的`addWaiter`函数

public final void acquire(int arg) {
        
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }
        
    //当前线程构造成Node放入CLH队列    
    //传入mode 为Node.EXCLUSIVE    
    private Node addWaiter(Node mode) {
        
    		//构造一个排它锁模式的Node节点
            Node node = new Node(Thread.currentThread(), mode);
            //快速路径。
            //由于enq函数插入CLH队列比较笨重，所以
            //所以写一个快速插入CLH队列方法。
            //尾节点作为下一个节点前节点
            Node pred = tail;
            //CLH头节点不为空(保证节点已经被初始化了)
            if (pred != null) {
        
            	//当前节点的前节点设置为上为尾结点
                node.prev = pred;
                //利用cas交换尾节点为当前插入节点
                if (compareAndSetTail(pred, node)) {
        
                	//尾节点的子节点是插入节点
                    pred.next = node;
                    return node;
                }
            }
            //快速路径失败，就老老实实的循环插入.
            //大多数情况，快速路径可以直接插入
            enq(node);
            return node;
        }
     private Node enq(final Node node) {
        
     		//利用 循环+cas 插入到成功为止 
            for (;;) {
        
    			
                Node t = tail;
                //为null 那么证明CLH没有初始化
                if (t == null) {
         // Must initialize
                	//创建一个空节点放入头部和尾部
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
        
                	//如果已经初始化，那么插入尾结点
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
        
                        t.next = node;
                        return t;
                    }
                }
            }
        }

简单理解就是  
(1) 若未进行初始化：设置一个空节点`Node`放入CLH头和尾结点。  
(2) 插入尾结点。

**初始化状态：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 3]  
**插入尾结点：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 4]

最后便是`acquireQueued`函数，`acquireQueued`用于自旋调用`tryAcquire`获得锁位置，当然会了降低空循环这里存在休眠操作。

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;
                    }
                    //shouldParkAfterFailedAcquire 判断当前是否需要休眠
                    //parkAndCheckInterrupt 休眠，并且检查中断，如果被中断返回true
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
        
            	//如果被中断那么取消获取锁资源清理操作
                if (failed)
                    cancelAcquire(node);
            }
        }

**shouldParkAfterFailedAcquire**：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        
            int ws = pred.waitStatus;
            if (ws == Node.SIGNAL)
                /*
                 * 该节点已经设置waitStatus为Node.SIGNAL，表示会释放一个信号给子节点，
                 * 所以当前线程可以安全的休眠，等候父节点唤醒。
                 */
                return true;
            if (ws > 0) {
        
                /*
                 * 
                 * 如果waitStatus>0 表示前一个节点被取消了，
                 * 所以我们直接，替换前节点到一个没有被取消的节点上
                 */
                do {
        
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else {
        
                /*
                 * 
                 * 结点Node的waitStatus肯定是0(未初始化状态)或者 PROPAGATE状态(可先忽视)，这两个状态
                 * 我们需要将其修改SIGNAL状态，但当前线程先不用休眠
                 * 
                 * 调用者需要重试以确保休眠之前不能获取到锁(重新循环).
                 */
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }

**parkAndCheckInterrupt**：  
最后是休眠函数，比较简单，休眠被唤醒后返回当前是否被中断。如果被中断返回true。

private final boolean parkAndCheckInterrupt() {
        
            LockSupport.park(this);
            return Thread.interrupted();
        }

为了不增加复杂度，关于**cancelAcquire**后续文章在做讲解。

上述便是加锁流程。

### 解锁流程 ###

//SimpleLock
     public void unLock() {
        
            release(1);
     }

释放

//AbstractQueuedSynchronizer
     public final boolean release(int arg) {
        
     		//表示自定义的释放锁函数
            if (tryRelease(arg)) {
        
                Node h = head;
                //如果存在头节点，并且waitStatus状态是不等于0
                //证明有子节点需要唤醒
                if (h != null && h.waitStatus != 0)
                	//唤醒子节点
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }

//SimpleLock
      @Override
        protected boolean tryRelease(int arg) {
        
        	//如果状态为0表示未被加锁
            if (getState() == 0) {
        
                throw new IllegalStateException();
            }
            //如果当前线程不是拥有锁对象
            if (getExclusiveOwnerThread() != Thread.currentThread()) {
        
                throw new IllegalStateException();
            }
            //清空逻辑
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

我们直接看最核心的函数`unparkSuccessor`

//AbstractQueuedSynchronizer
      private void unparkSuccessor(Node node) {
        
            /*
             * 如果当前节点的waitStatus小于可能他需要唤醒后续节点，
             * 所以这里把它重置为0，防止重复唤醒等操作。
             * 此处cas失败了不影响后续操作
             * 
             */
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
    
            /*
             * 唤醒当前节点的下一个子节点，如果子节点waitStatus为取消或者显示为null
             * 那么就继续向下找，当然Doug Lea 前辈这里根据实际的经验，从后往前找效率更高。
             * 因为从前往后找的话，有可能连续找到一连串的被取消的子节点。
             * 这里不用担心会出现间隔性出现取消的CLH队里，因为在插入的时候会
             * 在cancelAcquire和shouldParkAfterFailedAcquire处理，
             * 后续专门有文章讲解取消逻辑。
             */
            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);
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/50323e35693242d8bec478f2e33a9e57.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/a63ef4f1b3c74855a5ba655396f7d3e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/b465995c32734acab861978dc3683045.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/36f5544a0fd547f88e4c82a97badcb51.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/e57cfbf962b24f8a92deb9bf9c3c16ea.png