AQS原码剖析

小鱼儿 2024-04-17 15:33 25阅读 0赞

### 什么是线程安全问题？ ###

多个线程同时对共享资源进行操作，但并不能保证操作的原子性，可见性和有序性（在java中），由此会导致线程安全问题。

模拟一个抢票的场景，问题代码示例：

package lock;
    
    public class RobTicket {
        private static int ticket = 50;
    
        public static void main(String[] args) throws InterruptedException {
    
            for (int i = 1; i <= 50; i++) {
                new Thread(new Rob(i)).start();
            }
            Thread.sleep(1000);
        }
    
        static class Rob implements Runnable{
    
            private int number;
    
            public Rob(int i) {
                this.number = i;
            }
    
            @Override
            public void run() {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("线程-"+number+"-抢到票：" + ticket--);
            }
        }
    }

某一次结果如下：

线程-3-抢到票：49
    线程-2-抢到票：48
    线程-1-抢到票：49
    线程-5-抢到票：47
    线程-8-抢到票：44
    线程-7-抢到票：45
    ......

查看结果发现不同的线程有抢到同一张票

### 如何解决上述的线程安全问题？ ###

1.  使用AtomicInteger(volatile+cas)
2.  使用synchronized关键字
3.  使用lock

这里我使用ReentrantLock来解决这个问题

package lock;
    
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class RobTicket {
        private static int ticket = 50;
        private static Lock lock = new ReentrantLock();
    
        public static void main(String[] args) throws InterruptedException {
    
            for (int i = 1; i <= 50; i++) {
                new Thread(new Rob(i)).start();
            }
            Thread.sleep(1000);
        }
    
        static class Rob implements Runnable{
    
            private int number;
    
            public Rob(int i) {
                this.number = i;
            }
    
            @Override
            public void run() {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                lock.lock();
                try {
                    System.out.println("线程-"+number+"-抢到票：" + ticket--);
                } finally {
                    lock.unlock();
                }
            }
        }
    }

### ReentrantLock底层是如何实现的呢？答案是：AQS ###

剖析AQS之前，首先需要了解模板方法模式，可参考[设计模式之模板方法模式][Link 1]（重点是：模板方法和流程方法，下文会用到）

AQS在concurrent包中占据了半壁江山，很多并发工具类底层都是由AQS实现的。

### AQS中的流程方法 ###

*      tryAcquire
 *      tryRelease
 *      tryAcquireShared
 *      tryReleaseShared
 *      isHeldExclusively

这些方法在AQS中都是没有被实现的，如果需要实现独占锁，重写tryAcquire、tryRelease、isHeldExclusively方法即可；如果要实现共享锁，重写tryAcquireShared、tryReleaseShared、isHeldExclusively即可。先不关注AQS中的模板方法，我们通过重写流程方法来实现一个自己的锁，代码如下：

package lock;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.AbstractQueuedSynchronizer;
    import java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObject;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    
    public class MyLock implements Lock {
        private Sync sync = new Sync();
    
        @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();
        }
    
        private class Sync extends AbstractQueuedSynchronizer{
            @Override
            protected boolean tryAcquire(int arg) {
                int state = getState();
                if (state!=0){
                    return false;
                }
                if (compareAndSetState(0, 1)){
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            @Override
            protected boolean tryRelease(int arg) {
                setState(0);
                return true;
            }
    
            @Override
            protected boolean isHeldExclusively() {
                return getExclusiveOwnerThread() == Thread.currentThread();
            }
    
            private ConditionObject newCondition() {
                return new ConditionObject();
            }
        }
    }

看实现还是挺简单的，MyLock实现Lock接口，内部定义一个AQS的实现类Sync，重写tryAcquire、tryRelease、isHeldExclusively方法，newCondition是方便创建ConditionObject(AQS的非静态内部类)。测试一下MyLock能否解决现成安全为题。上代码：

package lock;
    
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class RobTicket3 {
        private static int ticket = 50;
        private static Lock lock = new MyLock();
    
        public static void main(String[] args) throws InterruptedException {
    
            for (int i = 1; i <= 50; i++) {
                new Thread(new Rob(i)).start();
            }
            Thread.sleep(1000);
        }
    
        static class Rob implements Runnable{
    
            private int number;
    
            public Rob(int i) {
                this.number = i;
            }
    
            public void run() {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                lock.lock();
                try {
                    System.out.println("线程-"+number+"-抢到票：" + ticket--);
                } finally {
                    lock.unlock();
                }
            }
        }
    }

线程-1-抢到票：50
    线程-4-抢到票：49
    线程-2-抢到票：48
    线程-5-抢到票：47
    线程-17-抢到票：46
    线程-3-抢到票：45
    线程-20-抢到票：44
    线程-15-抢到票：43
    线程-16-抢到票：42
    .......

通过查看结果，确定MyLock生效。下面应该深入到源码去看看AQS的模板方法的实现，不过在此之前先了解下当中的数据结构。

### AQS中的数据结构 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTQxMzYw_size_16_color_FFFFFF_t_70][]

如图所示为一个双向链表，每一个节点代表的是一个线程，另外有两个指针head、tail，分别指向了链表的头和尾，由此构成了一个同步器。当线程获取锁失败的时候，加入到同步队列的尾部，注意使用的CAS来设置的，而头结点并没有使用CAS。为什么呢？

**因为，当头结点获取到锁之后，出队，后一个节点成为新的头结点，不存在竞争。同一时刻可能有多个线程竞争锁失败然后需要加入到同步队列，所以需要使用CAS。**

### **AQS中的模板方法** ###

以accquire和release为例，上源码：

**先来看accquire的源码**

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

*      tryAcquire(int arg)

MyLock.Sync中重写了AQS的该方法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70][]

1.  获取得到state，0代表没有线程获取锁，1代表已有线程获取锁，已有线程获取锁的情况下直接返回失败
2.  尝试设置state的值为1，设置成功则获取锁成功，并设置锁被当前线程持有，返回成功

*      addWaiter(Node mode)

当线程尝试获取锁失败，讲当前线程加入到同步队列

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 1][]

1.  讲当前线程封装成一个节点
2.  尝试把当前线程设置为尾结点，成功则结束方法；失败则执行enq()方法
3.  看一下enq方法

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方法其实就是在一个死循环中不断地使用CAS来设置尾结点

*      acquireQueued(final Node node, int arg)

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 2][]

重点在for循环，他在这个死循环中的核心逻辑是：1、如果当前节点的上一个节点是头结点，则尝试获取锁，如果获取成功，则把当前节点设置为头结点；2、如果获取锁失败，则阻塞当前线程，知道其他线程释放锁

**再来看看release的源码**

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(arg)

MyLock.Sync中重写了AQS的tryRelease方法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 3][]

将state设置0即可

*      unparkSuccessor(Node node)

核心逻辑是唤醒后一个节点

致此，AQS中关于独占锁的代码解析完毕，共享锁是类似的，只是state的值不只是0和1，加锁是state++,解锁是state--，有兴趣的自己研究下，检验下学习成果^-^

### ReentrantLock中可重入锁和公平锁的概念 ###

之前在学习AQS的时候，实现了一个简版的MyLock，而concurrent包下的ReentrantLock中的实现要复杂许多，这里补充讲解下可重入和公平的概念在ReentrantLock中是如何实现的。

*  可重入锁：同一个线程如果多次获取同一把锁

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 4][]

这是独占锁的tryAcquire的实现，当中加入了一段逻辑（红框部分），如果锁已被持有，则判断持有者是否是当前线程，如果是，state++，返回获取锁成功

*  公平锁：先来的线程先获取锁

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 5][]

这是公平锁的tryAcquire的实现，当中新加入一段逻辑（红框部分），当锁没有被线程持有时，先判断同步队列中有线程存在吗，如果没有才尝试获取锁，以此保证了获取锁的公平性

[Link 1]: https://blog.csdn.net/qq_28411869/article/details/100115927
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTQxMzYw_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/9917c0a2a5a0431e937b656f5b8bfa3e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/e1949f2ae7d1485fb5fe7a5950e67875.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/a433701e3c524e1e9fef0e359494daba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/c87757e6d41a41e8b253f9d1f05658d9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/c9479d7f29fd4f338699ba9fb6c2ab93.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/9002aba678a141848b12b39519c2c208.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4NDExODY5_size_16_color_FFFFFF_t_70 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/ec9974c08eae48589c8eabdda4005f5d.png