AQS原理和介绍

起源

早期程序员会自己通过一种同步器去实现另一种相近的同步器，例如用可重入锁去实现信号量，或反之。这显然不
够优雅，于是在 JSR166（java 规范提案）中创建了 AQS，提供了这种通用的同步器机制。

目标

AQS 要实现的功能目标

• 阻塞版本获取锁 acquire 和非阻塞的版本尝试获取锁 tryAcquire
• 获取锁超时机制
• 通过打断取消机制
• 独占机制及共享机制
• 条件不满足时的等待机制

设计

AQS 的基本思想其实很简单
获取锁的逻辑

while(state 状态不允许获取) {
    
 if(队列中还没有此线程) {
    
 入队并阻塞
 }
}
当前线程出队

释放锁的逻辑

if(state 状态允许了) {
    
 恢复阻塞的线程(s) }

要点

• 原子维护 state 状态
• 阻塞及恢复线程
• 维护队列

1. state 设计

• state 使用 volatile 配合 cas 保证其修改时的原子性
• state 使用了 32bit int 来维护同步状态，因为当时使用 long 在很多平台下测试的结果并不理想

2. 阻塞恢复设计

• 早期的控制线程暂停和恢复的 api 有 suspend 和 resume，但它们是不可用的，因为如果先调用的 resume 那么 suspend 将感知不到
• 解决方法是使用 park & unpark 来实现线程的暂停和恢复，具体原理在之前讲过了，先 unpark 再 park 也没问题
• park & unpark 是针对线程的，而不是针对同步器的，因此控制粒度更为精细
• park 线程还可以通过 interrupt 打断

3. 队列设计

• 使用了 FIFO 先入先出队列，并不支持优先级队列
• 设计时借鉴了 CLH 队列，它是一种单向无锁队列

队列中有 head 和 tail 两个指针节点，都用 volatile 修饰配合 cas 使用，每个节点有 state 维护节点状态
入队伪代码，只需要考虑 tail 赋值的原子性

do {
    
 // 原来的 tail
 Node prev = tail;
 // 用 cas 在原来 tail 的基础上改为 node
} while(tail.compareAndSet(prev, node))

出队伪代码

// prev 是上一个节点
while((Node prev=node.prev).state != 唤醒状态) {
    
}
// 设置头节点
head = node;

CLH 好处：
无锁，使用自旋
快速，无阻塞
AQS 在一些方面改进了 CLH

    private Node enq(final Node node) {
    
        for (; ; ) {
    
            Node t = tail;
            // 队列中还没有元素 tail 为 null
            if (t == null) {
    
                // 将 head 从 null -> dummy
                if (compareAndSetHead(new Node())) {
    
                    tail = head;
                }
            } else {
    
                // 将 node 的 prev 设置为原来的 tail
                node.prev = t;
                // 将 tail 从原来的 tail 设置为 node
                if (compareAndSetTail(t, node)) {
    
                    // 原来 tail 的 next 设置为 node
                    t.next = node;
                    return t;
                }
            }
        }
    }

主要用到 AQS 的并发工具类

自定义锁的实现和测试

package com.example.demo.hmjuc.day19;

import com.example.demo.hmjuc.Sleep;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

@Slf4j
public class Test1 {
    

    public static void main(String[] args) {
    
        MyLock lock = new MyLock();
        new Thread(() -> {
    
            lock.lock();
            try {
    
                log.debug("locking...");
                Sleep.sleep(1000);
            } finally {
    
                log.debug("unlocking...");
                lock.unlock();
            }
        }, "t1").start();
        new Thread(() -> {
    
            lock.lock();
            try {
    
                log.debug("locking...");
            } finally {
    
                log.debug("unlocking...");
                lock.unlock();
            }
        }, "t2").start();
    }
}

/** * 自定义实现的锁 */
class MyLock implements Lock {
    
    /** * AQS锁 */
    private static final MySync SYNC = new MySync();

    /** * 加锁 * 成功就加锁 * 不成功 就进入队列 */
    @Override
    public void lock() {
    
        SYNC.acquire(1);
    }

    /** * 加锁可以被打断的 * * @throws InterruptedException 打断异常 */
    @Override
    public void lockInterruptibly() throws InterruptedException {
    
        SYNC.acquireInterruptibly(1);
    }

    /** * 尝试加锁 * 职场时一次 不会进入队列 * * @return true: 成功 */
    @Override
    public boolean tryLock() {
    
        return SYNC.tryAcquire(1);
    }

    /** * 尝试加锁 带超时时间 * * @param time 时间 * @param unit 时间单位 * @return true: 成功 * @throws InterruptedException 打断异常 */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    
        return SYNC.tryAcquireNanos(1, unit.toNanos(time));
    }

    /** * 解锁 */
    @Override
    public void unlock() {
    
        SYNC.release(1);
    }

    /** * 创建条件变量 * * @return 条件变量 */
    @Override
    public Condition newCondition() {
    
        return SYNC.newCondition();
    }

    /** * 自定义的同步工具 */
    static final class MySync extends AbstractQueuedSynchronizer {
    
        private static final long serialVersionUID = -3927501487421780249L;

        /** * 加锁 * 0 表示无锁 * 1 加锁 * 使用CAS 解决 * * @param age 年龄 * @return 加锁是否成功 */
        @Override
        protected boolean tryAcquire(int age) {
    
            // 加锁
            if (compareAndSetState(0, 1)) {
    
                // 设置当前线程持有锁
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        /** * 解锁 * * @param age 年龄 * @return 解锁是否成功 */
        @Override
        protected boolean tryRelease(int age) {
    
            // 当前线程已解锁
            if (getState() == 0) {
    
                throw new IllegalMonitorStateException();
            }
            // 设置持有锁的线程为空
            setExclusiveOwnerThread(null);
            // 设置无锁状态 注意: private volatile int state; 这里最后调用这个方法因为 这个 state变量是 volatile 修饰的可以
            // 保证代码不会被重排序
            setState(0);
            return true;
        }

        /** * 等待的条件 * * @return 条件 */
        private Condition newCondition() {
    
            return new ConditionObject();
        }

        /** * 判读当前线程是否持有锁 * * @return true: 持有, false: 没有持有 */
        @Override
        protected boolean isHeldExclusively() {
    
            return getState() == 1;
        }
    }
}

测试:

18:44:41.616 DEBUG    : locking...
18:44:42.623 DEBUG    : unlocking...
18:44:42.624 DEBUG    : locking...
18:44:42.624 DEBUG    : unlocking...