AQS 原理

概述

全称是 AbstractQueuedSynchronizer，是阻塞式锁和相关的同步器工具的框架

特点：

用 state 属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁
getState - 获取 state 状态

setState - 设置 state 状态

compareAndSetState - cas 机制设置 state 状态

独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源

提供了基于 FIFO 的等待队列，类似于 Monitor 的 EntryList
条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 Monitor 的 WaitSet

子类主要实现这样一些方法（默认抛出 UnsupportedOperationException）

tryAcquire
tryRelease
tryAcquireShared
tryReleaseShared
isHeldExclusively

获取锁的姿势

// 如果获取锁失败
if (!tryAcquire(arg)) {
 // 入队, 可以选择阻塞当前线程 park unpark
}

释放锁的姿势

// 如果释放锁成功
if (tryRelease(arg)) {
 // 让阻塞线程恢复运行
}

实现不可重入锁

自定义同步器

package schedule;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
//自定义锁(不可重入锁)
final class MySync extends AbstractQueuedSynchronizer {
   /** @param acquires 可重入数用来计数的,因为是不可重入锁，因此如果这里acquires的值大于一就返回false，表示加锁失败 */
    @Override
    protected boolean tryAcquire(int acquires) {
        if (acquires == 1) {
            //尝试加锁
            if (compareAndSetState(0, 1)) {
                //加上了锁,并设置owner为当前线程
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
        }
        //加锁失败
        return false;
    }
    /** @param acquires 可重入数用来计数的,因为是不可重入锁，因此如果这里acquires的值大于一就返回false，表示解锁失败 */
    @Override
    protected boolean tryRelease(int acquires) {
        if (acquires == 1) {
            //当前为没有上锁的状态
            if (getState() == 0) {
                throw new IllegalMonitorStateException();
            }
            //设置当前没有线程占用锁
            setExclusiveOwnerThread(null);
            //解锁--- private volatile int state
            //因为state是volatile,因此将setState(0);方法执行放在setExclusiveOwnerThread(null);方法执行之前可以确保
            //不会产生指令重排; 确保线程可见性;----exclusiveOwnerThread不是volatile
            setState(0);
            return true;
        }
        return false;
    }
    //条件变量
    protected Condition newCondition() {
        return new ConditionObject();
    }
    //是否持有独占锁--1:持有; 0不持有
    @Override
    protected boolean isHeldExclusively() {
        return getState() == 1;
    }
}

同步器类中的大部分方法由其父类提供

自定义锁

有了自定义同步器，很容易复用 AQS ，实现一个功能完备的自定义锁

package schedule;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
class MyLock implements Lock {
 static MySync sync = new MySync();
 @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();
 }
}

release和tryRelease的区别:

测试一下

MyLock lock = new MyLock();
        new Thread(() -> {
            lock.lock();
            try {
                log.debug("locking...");
                sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                log.debug("unlocking...");
                lock.unlock();
            }
        }, "t1").start();
        new Thread(() -> {
            lock.lock();
            try {
                log.debug("locking...");
            } finally {
                log.debug("unlocking...");
                lock.unlock();
            }
        }, "t2").start();

输出

22:29:28.727 c.TestAqs [t1] - locking... 
22:29:29.732 c.TestAqs [t1] - unlocking... 
22:29:29.732 c.TestAqs [t2] - locking... 
22:29:29.732 c.TestAqs [t2] - unlocking...

不可重入测试

如果改为下面代码，会发现自己也会被挡住（只会打印一次 locking）

lock.lock();
log.debug("locking...");
lock.lock();
log.debug("locking...");

心得

起源

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

目标

AQS 要实现的功能目标

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

要实现的性能目标
Instead, the primary performance goal here is scalability: topredictably maintain efficiency even, or especially, when
synchronizers are contended.

设计

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 的并发工具类

ReentrantLock 原理

非公平锁实现原理

加锁流程

先从构造器开始看，默认为非公平锁实现

public ReentrantLock() {
 sync = new NonfairSync();
}

NonfairSync 继承自 AQS

没有竞争,加锁成功的情况:

第一个竞争出现时

Thread-1 执行了

CAS 尝试将 state 由 0 改为 1，结果失败
进入 tryAcquire 逻辑，这时 state 已经是1，结果仍然失败
接下来进入 addWaiter 逻辑，构造 Node 队列

图中黄色三角表示该 Node 的 waitStatus 状态，其中 0 为默认正常状态
Node 的创建是懒惰的
其中第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程

当前线程进入 acquireQueued 逻辑
4. acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 park 阻塞
5. 如果自己是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
6. 进入 shouldParkAfterFailedAcquire 逻辑，将前驱 node，即 head 的 waitStatus 改为 -1，这次返回 false

shouldParkAfterFailedAcquire 执行完毕回到 acquireQueued ，再次 tryAcquire 尝试获取锁，当然这时
state 仍为 1，失败

8.当再次进入 shouldParkAfterFailedAcquire 时，这时因为其前驱 node 的 waitStatus 已经是 -1，这次返回
true

9.进入 parkAndCheckInterrupt， Thread-1 park（灰色表示）

再次有多个线程经历上述过程竞争失败，变成这个样子

解锁流程

Thread-0 释放锁，进入 tryRelease 流程，如果成功

设置 exclusiveOwnerThread 为 null
state = 0

当前队列不为 null，并且 head 的 waitStatus = -1，进入 unparkSuccessor 流程—唤醒后继节点
找到队列中离 head 最近的一个 Node（没取消的），unpark 恢复其运行，本例中即为 Thread-1
回到 Thread-1 的 acquireQueued 流程

如果加锁成功（没有竞争），会设置

exclusiveOwnerThread 为 Thread-1，state = 1
head 指向刚刚 Thread-1 所在的 Node，该 Node 清空 Thread
原本的 head 因为从链表断开，而可被垃圾回收

如果这时候有其它线程来竞争（非公平的体现），例如这时有 Thread-4 来了

如果不巧又被 Thread-4 占了先

Thread-4 被设置为 exclusiveOwnerThread，state = 1
Thread-1 再次进入 acquireQueued 流程，获取锁失败，重新进入 park 阻塞

加锁流程源码分析:

加锁源码

// Sync 继承自 AQS
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;
        // 加锁实现
        final void lock() {
            // 首先用 cas 尝试（仅尝试一次）将 state 从 0 改为 1, 如果成功表示获得了独占锁
            if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread());
            else
                // 如果尝试失败，进入 ㈠
                acquire(1);
        }
        // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处
        public final void acquire(int arg) {
            // ㈡ tryAcquire 
            if (!tryAcquire(arg) &&
                    // 当 tryAcquire 返回为 false 时, 先调用 addWaiter ㈣, 接着 acquireQueued ㈤
                    acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {
                selfInterrupt();
            }
        }
        // ㈡ 进入 ㈢
        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
        // ㈢ Sync 继承过来的方法, 方便阅读, 放在此处
        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            // 如果还没有获得锁
            if (c == 0) {
                // 尝试用 cas 获得, 这里体现了非公平性: 不去检查 AQS 队列
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            // 如果已经获得了锁, 线程还是当前线程, 表示发生了锁重入
            else if (current == getExclusiveOwnerThread()) {
                // state++
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            // 获取失败, 回到调用处
            return false;
        }
        // ㈣ AQS 继承过来的方法, 方便阅读, 放在此处
        private Node addWaiter(Node mode) {
            // 将当前线程关联到一个 Node 对象上, 模式为独占模式
            Node node = new Node(Thread.currentThread(), mode);
            // 如果 tail 不为 null, cas 尝试将 Node 对象加入 AQS 队列尾部
            Node pred = tail;
            if (pred != null) {
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                    // 双向链表
                    pred.next = node;
                    return node;
                }
            }
            // 尝试将 Node 加入 AQS, 进入 ㈥
            enq(node);
            return node;
        }
        // ㈥ AQS 继承过来的方法, 方便阅读, 放在此处
        private Node enq(final Node node) {
            for (; ; ) {
                Node t = tail;
                if (t == null) {
                    // 还没有, 设置 head 为哨兵节点（不对应线程，状态为 0）
                    if (compareAndSetHead(new Node())) {
                        tail = head;
                    }
                } else {
                    // cas 尝试将 Node 对象加入 AQS 队列尾部
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }
        // ㈤ AQS 继承过来的方法, 方便阅读, 放在此处
        final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (; ; ) {
                    final Node p = node.predecessor();
                    // 上一个节点是 head, 表示轮到自己（当前线程对应的 node）了, 尝试获取
                    if (p == head && tryAcquire(arg)) {
                        // 获取成功, 设置自己（当前线程对应的 node）为 head
                        setHead(node);
                        // 上一个节点 help GC
                        p.next = null;
                        failed = false;
                        // 返回中断标记 false
                        return interrupted;
                    }
                    if (
                        // 判断是否应当 park, 进入 ㈦
                            shouldParkAfterFailedAcquire(p, node) &&
                                    // park 等待, 此时 Node 的状态被置为 Node.SIGNAL ㈧
                                    parkAndCheckInterrupt()) {
                        interrupted = true;
                    }
                }
            } finally {
                if (failed) cancelAcquire(node);
            }
        }
        // ㈦ AQS 继承过来的方法, 方便阅读, 放在此处
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
            // 获取上一个节点的状态
            int ws = pred.waitStatus;
            if (ws == Node.SIGNAL) {
                // 上一个节点都在阻塞, 那么自己也阻塞好了
                return true;
            }
            // > 0 表示取消状态
            if (ws > 0) {
                // 上一个节点取消, 那么重构删除前面所有取消的节点, 返回到外层循环重试
                do {
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else {
                // 这次还没有阻塞
                // 但下次如果重试不成功, 则需要阻塞，这时需要设置上一个节点状态为 Node.SIGNAL
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }
        // ㈧ 阻塞当前线程
        private final boolean parkAndCheckInterrupt() {
            LockSupport.park(this);
            return Thread.interrupted();
        }
    }

注意

是否需要 unpark 是由当前节点的前驱节点的 waitStatus == Node.SIGNAL 来决定，而不是本节点的 waitStatus 决定

解锁流程

解锁不区分公平和非公平

被唤醒的线程，恢复到最初被park阻塞的位置:

源码:

// Sync 继承自 AQS
    static final class NonfairSync extends Sync {
        // 解锁实现
        public void unlock() {
            sync.release(1);
        }
        // AQS 继承过来的方法, 方便阅读, 放在此处
        public final boolean release(int arg) {
            // 尝试释放锁, 进入 ㈠
            if (tryRelease(arg)) {
                // 队列头节点 unpark
                Node h = head;
                if (
                    // 队列不为 null
                        h != null &&
                                // waitStatus == Node.SIGNAL 才需要 unpark
                                h.waitStatus != 0) {
                    // unpark AQS 中等待的线程, 进入 ㈡
                    unparkSuccessor(h);
                }
                return true;
            }
            return false;
        }
        // ㈠ Sync 继承过来的方法, 方便阅读, 放在此处
        protected final boolean tryRelease(int releases) {
            // state--
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException();
            boolean free = false;
            // 支持锁重入, 只有 state 减为 0, 才释放成功
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }
        // ㈡ AQS 继承过来的方法, 方便阅读, 放在此处
        private void unparkSuccessor(Node node) {
            // 如果状态为 Node.SIGNAL 尝试重置状态为 0
            // 不成功也可以
            int ws = node.waitStatus;
            if (ws < 0) {
                compareAndSetWaitStatus(node, ws, 0);
            }
            // 找到需要 unpark 的节点, 但本节点从 AQS 队列中脱离, 是由唤醒节点完成的
            Node s = node.next;
            // 不考虑已取消的节点, 从 AQS 队列从后至前找到队列最前面需要 unpark 的节点
            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);
        }
    }

可重入原理

static final class NonfairSync extends Sync {
        // ...
        // Sync 继承过来的方法, 方便阅读, 放在此处
        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            // 如果已经获得了锁, 线程还是当前线程, 表示发生了锁重入
            else if (current == getExclusiveOwnerThread()) {
                // state++
                //如果此时c=1,表示当前线程已经拥有了这把锁，但是此时出现了锁重入的情况，state计数+1=2
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
        // Sync 继承过来的方法, 方便阅读, 放在此处
        protected final boolean tryRelease(int releases) {
            // state--
            //当前线程state计数减去一
            int c = getState() - releases;
            //如果当前线程没有用有锁，还调用释放锁的方法会抛出异常
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
                //free变量标记是否释放了锁
            boolean free = false;
            // 支持锁重入, 只有 state 减为 0, 才释放成功
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }
    }

可打断原理

不可打断模式

在此模式下，即使它被打断，仍会驻留在 AQS 队列中，一直要等到获得锁后方能得知自己被打断了

// Sync 继承自 AQS
    static final class NonfairSync extends Sync {
        // ...
        private final boolean parkAndCheckInterrupt() {
            // 如果打断标记已经是 true, 则 park 会失效
            LockSupport.park(this);
            // interrupted 会清除打断标记
            return Thread.interrupted();
        }
        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;
                        failed = false;
                        // 还是需要获得锁后, 才能返回打断状态
                        return interrupted;
                    }
                    if (
                            shouldParkAfterFailedAcquire(p, node) &&
                            //parkAndCheckInterrupt方法在被唤醒或打断后，返回的是当前线程的打断标记，即打断状态
                            //如果被打断了返回true，会进入if语句设置interrupted的值为true，再次尝试去获取锁
                                    parkAndCheckInterrupt()
                    ) {
                        // 如果是因为 interrupt 被唤醒, 返回打断状态为 true
                        interrupted = true;
                    }
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
        public final void acquire(int arg) {
            if (
                    !tryAcquire(arg) &&
                            acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
            ) {
                // 如果打断状态为 true
                selfInterrupt();
            }
        }
        static void selfInterrupt() {
            // 重新产生一次中断
            Thread.currentThread().interrupt();
        }
    }

可打断模式

static final class NonfairSync extends Sync {
        public final void acquireInterruptibly(int arg) throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            // 如果没有获得到锁, 进入 ㈠
            if (!tryAcquire(arg))
                doAcquireInterruptibly(arg);
        }
        // ㈠ 可打断的获取锁流程
        private void doAcquireInterruptibly(int arg) throws InterruptedException {
            final Node node = addWaiter(Node.EXCLUSIVE);
            boolean failed = true;
            try {
                for (; ; ) {
                    final Node p = node.predecessor();
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                    // //parkAndCheckInterrupt方法在被唤醒或打断后，返回的是当前线程的打断标记，即打断状态
                            //如果被打断了返回true，会进入if语句,然后抛出被打断的异常
                            parkAndCheckInterrupt()) {
                        // 在 park 过程中如果被 interrupt 会进入此
                        // 这时候抛出异常, 而不会再次进入 for (;;)
                        throw new InterruptedException();
                    }
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    }

公平锁实现原理

static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;
        final void lock() {
            acquire(1);
        }
        // AQS 继承过来的方法, 方便阅读, 放在此处
        public final void acquire(int arg) {
            if (
                    !tryAcquire(arg) &&
                            acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
            ) {
                selfInterrupt();
            }
        }
        // 与非公平锁主要区别在于 tryAcquire 方法的实现
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                // 先检查 AQS 队列中是否有前驱节点, 没有才去竞争
                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;
        }
        // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处
        public final boolean hasQueuedPredecessors() {
            Node t = tail;
            Node h = head;
            Node s;
            // h != t 时表示队列中有 Node
            return h != t &&
                    (
                            // (s = h.next) == null 表示队列中还有没有老二
                            //头结点是哑元，老二节点优先级最高
                            (s = h.next) == null ||
                                    // 或者队列中老二线程不是此线程
                                    s.thread != Thread.currentThread()
                    );
        }
    }

对比非公平锁的实现

// ㈢ Sync 继承过来的方法, 方便阅读, 放在此处
        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            // 如果还没有获得锁
            if (c == 0) {
                // 尝试用 cas 获得, 这里体现了非公平性: 不去检查 AQS 队列
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            // 如果已经获得了锁, 线程还是当前线程, 表示发生了锁重入
            else if (current == getExclusiveOwnerThread()) {
                // state++
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            // 获取失败, 回到调用处
            return false;
        }

条件变量实现原理

每个条件变量其实就对应着一个等待队列，其实现类是 ConditionObject

await 流程

开始 Thread-0 持有锁，调用 await，进入 ConditionObject 的 addConditionWaiter 流程

创建新的 Node 状态为 -2（Node.CONDITION），关联 Thread-0，加入等待队列尾部

await()对应源码

public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            //创建新的Node状态为-2（Node.CONDITION），关联 Thread-0，加入等待队列尾部
            Node node = addConditionWaiter();
             .....
        }

addConditionWaiter()对应源码

//将等待者加入等待队列，返回新创建的Node等待节点
    private Node addConditionWaiter() 
    {
            Node t = lastWaiter;
            // If lastWaiter is cancelled, clean out.
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
            //新创建的节点关联当前线程，值为-2(Node.CONDITION值为-2)
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            //如果等待队列没有节点，那么当前新创建出来的节点作为头部
            //否则作为队列尾部节点
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }

接下来进入 AQS 的 fullyRelease 流程，释放同步器上的锁

unpark AQS 队列中的下一个节点，竞争锁，假设没有其他竞争线程，那么 Thread-1 竞争成功

await()对应源码

public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            //创建新的Node状态为-2（Node.CONDITION），关联 Thread-0，加入等待队列尾部
            Node node = addConditionWaiter();
            //进入fullyRelease(node)释放锁的流程
            int savedState = fullyRelease(node);
            ....
        }

fullyRelease(node)对应源码

final int fullyRelease(Node node) {
        boolean failed = true;
        try {
           //获取当前state的值
            int savedState = getState();
            //释放锁，state值清零，即释放所有锁(可重入锁state值大于一)
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }

release(savedState)对应源码

public final boolean release(int arg) 
    {       
    //释放锁成功
        if (tryRelease(arg)) 
        {
            Node h = head;
              //头结点不为空，并且waitStatus 为-1
            if (h != null && h.waitStatus != 0)
            //唤醒头结点的后继节点--unpark
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

tryRelease(arg)对应源码

protected final boolean tryRelease(int releases) 
   {        
            //这里releases的值为int savedState = getState();
            //因此这里c为0，相当于释放了当前线程加的所有锁(可重入锁)
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            //锁被释放了
            if (c == 0)
             {
                free = true;
                //设置owner值为null
                setExclusiveOwnerThread(null);
            }
            //解锁
            setState(c);
            return free;
        }

park 阻塞 Thread-0

await()对应源码

public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            //创建新的Node状态为-2（Node.CONDITION），关联 Thread-0，加入等待队列尾部
            Node node = addConditionWaiter();
            //进入fullyRelease(node)释放锁的流程
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                //调用park(this)方法,将自己阻塞，进入休息室等待
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
             ...
        }

signal 流程

假设 Thread-1 要来唤醒 Thread-0

signal()源码

public final void signal() {
            //当前线程是否是锁的持有者
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            //获取等待队列中第一个节点 
            Node first = firstWaiter;
            if (first != null)
            //如果第一个节点不为空，就唤醒第一个节点
            //这里可以看出，底层不是随机唤醒
                doSignal(first);
        }

进入 ConditionObject 的 doSignal 流程，取得等待队列中第一个 Node，即 Thread-0 所在 Node

执行 transferForSignal 流程，将该 Node 加入 AQS 队列尾部，将 Thread-0 的 waitStatus 改为 0，Thread-3 的
waitStatus 改为 -1

Thread-1 释放锁，进入 unlock 流程，略
doSignal()源码

private void doSignal(Node first) {
            do {
            //将当前等待队列中的第一个节点，从等待队列的双向链表中断开连接
            //如果第一个节点有后继节点，那么让后继节点成为第一个节点
                if ( (firstWaiter = first.nextWaiter) == null)
                //如果没有后继节点，说明当前队列就一个元素,lastWaiter 指针值变为null
                    lastWaiter = null;
                    //从链表中断开连接
                first.nextWaiter = null;
                //transferForSignal尝试将被唤醒的节点，加入AQS阻塞队列尾部
                //这里可能会操作失败，因为在加入队列过程中，存在被打断，取消的可能
            } while (!transferForSignal(first) &&
            //如果加入队列尾部失败，就再尝试去唤醒等待队列中下一个节点
                     (first = firstWaiter) != null);
        }

transferForSignal(first)源码

/** * 将当前节点从等待队列转移到同步队列 * 转移成功返回true * 如果转移过程中当前节点被取消了，那么转移失败 */
    final boolean transferForSignal(Node node) {
        /* * 将当前节点的状态由-2切换为0，如果切换失败，表示当前节点被取消了 */
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        /* * 将被唤醒的节点加入同步队列的尾部，如果加入成功，返回的是加入成功后该节点的前驱节点 */
        Node p = enq(node);
        //获取前驱节点的状态
        int ws = p.waitStatus;
        //设置前驱节点的状态为-1，表示他有职责唤醒他的后继节点
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        //如果转移失败了，会唤醒该节点的线程
            LockSupport.unpark(node.thread);
        return true;
    }

ConditionObject 类源码

public class ConditionObject implements Condition, java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    // 第一个等待节点
    private transient Node firstWaiter;
    // 最后一个等待节点
    private transient Node lastWaiter;
    public ConditionObject() {
    }
    // ㈠ 添加一个 Node 至等待队列
    private Node addConditionWaiter() {
        Node t = lastWaiter;
        // 所有已取消的 Node 从队列链表删除, 见 ㈡
        if (t != null && t.waitStatus != Node.CONDITION) {
            unlinkCancelledWaiters();
            t = lastWaiter;
        }
        // 创建一个关联当前线程的新 Node, 添加至队列尾部
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        if (t == null)
            firstWaiter = node;
        else
            t.nextWaiter = node;
        lastWaiter = node;
        return node;
    }
    // 唤醒 - 将没取消的第一个节点转移至 AQS 队列
    private void doSignal(Node first) {
        do {
            // 已经是尾节点了
            if ((firstWaiter = first.nextWaiter) == null) {
                lastWaiter = null;
            }
            first.nextWaiter = null;
        } while (
            // 将等待队列中的 Node 转移至 AQS 队列, 不成功且还有节点则继续循环 ㈢
                !transferForSignal(first) &&
                        // 队列还有节点
                        (first = firstWaiter) != null
        );
    }
    // 外部类方法, 方便阅读, 放在此处
    // ㈢ 如果节点状态是取消, 返回 false 表示转移失败, 否则转移成功
    final boolean transferForSignal(Node node) {
        // 如果状态已经不是 Node.CONDITION, 说明被取消了
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        // 加入 AQS 队列尾部
        Node p = enq(node);
        int ws = p.waitStatus;
        if (
            // 上一个节点被取消
                ws > 0 ||
                        // 上一个节点不能设置状态为 Node.SIGNAL
                        !compareAndSetWaitStatus(p, ws, Node.SIGNAL)
        ) {
            // unpark 取消阻塞, 让线程重新同步状态
            LockSupport.unpark(node.thread);
        }
        return true;
    }
    // 全部唤醒 - 等待队列的所有节点转移至 AQS 队列
    private void doSignalAll(Node first) {
        lastWaiter = firstWaiter = null;
        do {
            Node next = first.nextWaiter;
            first.nextWaiter = null;
            transferForSignal(first);
            first = next;
        } while (first != null);
    }
    // ㈡
    private void unlinkCancelledWaiters() {
        // ...
    }
    // 唤醒 - 必须持有锁才能唤醒, 因此 doSignal 内无需考虑加锁
    public final void signal() {
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
            doSignal(first);
    }
    // 全部唤醒 - 必须持有锁才能唤醒, 因此 doSignalAll 内无需考虑加锁
    public final void signalAll() {
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
            doSignalAll(first);
    }
    // 不可打断等待 - 直到被唤醒
    public final void awaitUninterruptibly() {
        // 添加一个 Node 至等待队列, 见 ㈠
        Node node = addConditionWaiter();
        // 释放节点持有的锁, 见 ㈣
        int savedState = fullyRelease(node);
        boolean interrupted = false;
        // 如果该节点还没有转移至 AQS 队列, 阻塞
        while (!isOnSyncQueue(node)) {
            // park 阻塞
            LockSupport.park(this);
            // 如果被打断, 仅设置打断状态
            if (Thread.interrupted())
                interrupted = true;
        }
        // 唤醒后, 尝试竞争锁, 如果失败进入 AQS 队列
        if (acquireQueued(node, savedState) || interrupted)
            selfInterrupt();
    }
    // 外部类方法, 方便阅读, 放在此处
    // ㈣ 因为某线程可能重入，需要将 state 全部释放
    final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            int savedState = getState();
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }
    // 打断模式 - 在退出等待时重新设置打断状态
    private static final int REINTERRUPT = 1;
    // 打断模式 - 在退出等待时抛出异常
    private static final int THROW_IE = -1;
    // 判断打断模式
    private int checkInterruptWhileWaiting(Node node) {
        return Thread.interrupted() ?
                (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                0;
    }
    // ㈤ 应用打断模式
    private void reportInterruptAfterWait(int interruptMode)
            throws InterruptedException {
        if (interruptMode == THROW_IE)
            throw new InterruptedException();
        else if (interruptMode == REINTERRUPT)
            selfInterrupt();
    }
    // 等待 - 直到被唤醒或打断
    public final void await() throws InterruptedException {
        if (Thread.interrupted()) {
            throw new InterruptedException();
        }
        // 添加一个 Node 至等待队列, 见 ㈠
        Node node = addConditionWaiter();
        // 释放节点持有的锁
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        // 如果该节点还没有转移至 AQS 队列, 阻塞
        while (!isOnSyncQueue(node)) {
            // park 阻塞
            LockSupport.park(this);
            // 如果被打断, 退出等待队列
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        // 退出等待队列后, 还需要获得 AQS 队列的锁
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        // 所有已取消的 Node 从队列链表删除, 见 ㈡
        if (node.nextWaiter != null)
            unlinkCancelledWaiters();
        // 应用打断模式, 见 ㈤
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }
    // 等待 - 直到被唤醒或打断或超时
    public final long awaitNanos(long nanosTimeout) throws InterruptedException {
        if (Thread.interrupted()) {
            throw new InterruptedException();
        }
        // 添加一个 Node 至等待队列, 见 ㈠
        Node node = addConditionWaiter();
        // 释放节点持有的锁
        int savedState = fullyRelease(node);
        // 获得最后期限
        final long deadline = System.nanoTime() + nanosTimeout;
        int interruptMode = 0;
        // 如果该节点还没有转移至 AQS 队列, 阻塞
        while (!isOnSyncQueue(node)) {
            // 已超时, 退出等待队列
            if (nanosTimeout <= 0L) {
                transferAfterCancelledWait(node);
                break;
            }
            // park 阻塞一定时间, spinForTimeoutThreshold 为 1000 ns
            if (nanosTimeout >= spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);
            // 如果被打断, 退出等待队列
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
            nanosTimeout = deadline - System.nanoTime();
        }
        // 退出等待队列后, 还需要获得 AQS 队列的锁
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        // 所有已取消的 Node 从队列链表删除, 见 ㈡
        if (node.nextWaiter != null)
            unlinkCancelledWaiters();
        // 应用打断模式, 见 ㈤
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
        return deadline - System.nanoTime();
    }
    // 等待 - 直到被唤醒或打断或超时, 逻辑类似于 awaitNanos
    public final boolean awaitUntil(Date deadline) throws InterruptedException {
        // ...
    }
    // 等待 - 直到被唤醒或打断或超时, 逻辑类似于 awaitNanos
    public final boolean await(long time, TimeUnit unit) throws InterruptedException {
        // ...
    }
    // 工具方法 省略 ...
}

读写锁原理

ReentrantReadWriteLock

当读操作远远高于写操作时，这时候使用读写锁让 读-读 可以并发，提高性能。类似于数据库中的 select ...from ... lock in share mode

提供一个 数据容器类 内部分别使用读锁保护数据的 read() 方法，写锁保护数据的 write() 方法

package reentrantLock;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static java.lang.Thread.sleep;
@Slf4j
class DataContainer {
    private Object data;
    private ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
    private ReentrantReadWriteLock.ReadLock r = rw.readLock();
    private ReentrantReadWriteLock.WriteLock w = rw.writeLock();
    @SneakyThrows
    public Object read() {
        log.debug("获取读锁...");
        r.lock();
        try {
            log.debug("读取");
            sleep(1);
            return data;
        } finally {
            log.debug("释放读锁...");
            r.unlock();
        }
    }
    @SneakyThrows
    public void write() {
        log.debug("获取写锁...");
        w.lock();
        try {
            log.debug("写入");
            sleep(1);
        } finally {
            log.debug("释放写锁...");
            w.unlock();
        }
    }
}

测试 读锁-读锁 可以并发

package reentrantLock;
/** * @author 大忽悠 * @create 2022/1/7 13:26 */
public class Main
{
    public static void main(String[] args)
    {
        DataContainer dataContainer = new DataContainer();
        new Thread(() -> {
            dataContainer.read();
        }, "t1").start();
        new Thread(() -> {
            dataContainer.read();
        }, "t2").start();
    }
}

输出结果，从这里可以看到 Thread-0 锁定期间，Thread-1 的读操作不受影响

14:05:14.341 c.DataContainer [t2] - 获取读锁... 
14:05:14.341 c.DataContainer [t1] - 获取读锁... 
14:05:14.345 c.DataContainer [t1] - 读取
14:05:14.345 c.DataContainer [t2] - 读取
14:05:15.365 c.DataContainer [t2] - 释放读锁... 
14:05:15.386 c.DataContainer [t1] - 释放读锁...

测试 读锁-写锁 相互阻塞

在这里插入代码片

输出结果

14:04:21.838 c.DataContainer [t1] - 获取读锁... 
14:04:21.838 c.DataContainer [t2] - 获取写锁... 
14:04:21.841 c.DataContainer [t2] - 写入
14:04:22.843 c.DataContainer [t2] - 释放写锁... 
14:04:22.843 c.DataContainer [t1] - 读取
14:04:23.843 c.DataContainer [t1] - 释放读锁...

写锁-写锁 也是相互阻塞的，这里就不测试了

注意事项

读锁不支持条件变量
重入时升级不支持：即持有读锁的情况下去获取写锁，会导致获取写锁永久等待

r.lock();
        try {
            // ...
            w.lock();
            try {
                // ...
            } finally {
                w.unlock();
            }
        } finally {
            r.unlock();
        }

重入时降级支持：即持有写锁的情况下去获取读锁

package reentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
class CachedData<ReentrantReadWriteLock> {
    Object data;
    // 是否有效，如果失效，需要重新计算 data
    volatile boolean cacheValid;
    final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    void processCachedData() {
        rwl.readLock().lock();
        if (!cacheValid) {
            // 获取写锁前必须释放读锁
            rwl.readLock().unlock();
            rwl.writeLock().lock();
            try {
                // 判断是否有其它线程已经获取了写锁、更新了缓存, 避免重复更新
                if (!cacheValid) {
                    data = ...
                    cacheValid = true;
                }
                // 降级为读锁, 释放写锁, 这样能够让其它线程读取缓存
                rwl.readLock().lock();
            } finally {
                rwl.writeLock().unlock();
            }
        }
        // 自己用完数据, 释放读锁 
        try {
            use(data);
        } finally {
            rwl.readLock().unlock();
        }
    }
}

缓存

1. 缓存更新策略

更新时，是先清缓存还是先更新数据库

先清缓存

先更新数据库

补充一种情况，假设查询线程 A 查询数据时恰好缓存数据由于时间到期失效，或是第一次查询

这种情况的出现几率非常小，见 facebook 论文

2. 读写锁实现一致性缓存

使用读写锁实现一个简单的按需加载缓存

package com;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Objects;
import java.util.concurrent.locks.ReentrantReadWriteLock;
class GenericCachedDao<T> {
    /** * HashMap 作为缓存非线程安全, 需要保护 */
    HashMap<SqlPair, T> map = new HashMap<>();
    /** * 读写锁 */
    ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    /** * 数据库查询类 */
    GenericDao genericDao = new GenericDao();
    /** * <p> * 更新方法 * </p> * @param sql * @param params * @return */
    public int update(String sql, Object... params) {
        SqlPair key = new SqlPair(sql, params);
        // 加写锁, 防止其它线程对缓存读取和更改
        lock.writeLock().lock();
        try {
            //先更新数据库
            int rows = genericDao.update(sql, params);
            //再清空缓存
            map.clear();
            return rows;
        } finally {
            //释放写锁
            lock.writeLock().unlock();
        }
    }
    /** * @param beanClass * @param sql * @param params * @return */
    public T queryOne(Class<T> beanClass, String sql, Object... params) {
        SqlPair key = new SqlPair(sql, params);
        // 加读锁, 防止其它线程对缓存更改
        lock.readLock().lock();
        try {
            T value = map.get(key);
            if (value != null) {
                return value;
            }
        } finally {
            lock.readLock().unlock();
        }
        // 加写锁, 防止其它线程对缓存读取和更改
        lock.writeLock().lock();
        try {
            // get 方法上面部分是可能多个线程进来的, 可能已经向缓存填充了数据
            // 为防止重复查询数据库, 再次验证
            T value = map.get(key);
            if (value == null) {
                // 如果没有, 查询数据库
                value = genericDao.queryOne(beanClass, sql, params);
                map.put(key, value);
            }
            return value;
        } finally {
            lock.writeLock().unlock();
        }
    }
    // 作为 key 保证其是不可变的
    class SqlPair
    {
        /** * sql */
        private String sql;
        /** * 参数 */
        private Object[] params;
        public SqlPair(String sql, Object[] params) {
            this.sql = sql;
            this.params = params;
        }
        /** * @param o * @return 这里需要重写equals和hashcode方法，因为默认是根据对象地址进行判断的 */
        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            SqlPair sqlPair = (SqlPair) o;
            return sql.equals(sqlPair.sql) && Arrays.equals(params, sqlPair.params);
        }
        /** * @return 这里的hashcode计算值是以内部属性计算得出的 */
        @Override
        public int hashCode() {
            int result = Objects.hash(sql);
            result = 31 * result + Arrays.hashCode(params);
            return result;
        }
    }
}

注意

以上实现体现的是读写锁的应用，保证缓存和数据库的一致性，但有下面的问题没有考虑

适合读多写少，如果写操作比较频繁，以上实现性能低
没有考虑缓存容量
没有考虑缓存过期
只适合单机
并发性还是低，目前只会用一把锁,例如读取表一的同时，需要对表二进行写操作，本是两个互不相关联的操作，却因为用了同一把锁，而会产生阻塞现象,可以尝试将锁进行细粒度划分，例如一个表一把锁
更新方法太过简单粗暴，清空了所有 key（考虑按类型分区或重新设计 key）,比如，我对表1进行的更新，却把表一和表二的缓存的清空了，这显然不合理

乐观锁实现：用 CAS 去更新

读写锁原理

1. 图解流程

加锁

读写锁用的是同一个 Sycn 同步器，因此等待队列、state 等也是同一个

t1 w.lock，t2 r.lock

1） t1 成功上锁，流程与 ReentrantLock 加锁相比没有特殊之处，不同是写锁状态占了 state 的低 16 位，而读锁使用的是 state 的高 16 位

tryAcquire(int acquires)源码

protected final boolean tryAcquire(int acquires) {
            Thread current = Thread.currentThread();
            //获取state的值
            int c = getState();
            //获取写锁部分
            int w = exclusiveCount(c);
            //c!=0表示有人加锁了，但是这里高十六位或者第十六位不等于0都会导致c不等于0，无法判断是加了读锁还是写锁 
            if (c != 0) {
                // (Note: if c != 0 and w == 0 then shared count != 0)
                //如果写锁部分等于0，表示加的是读锁,读锁和写锁是互斥的，因此这里返回false
                //判断写锁是不是自己加的,如果是被人加的，返回false
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                    //写锁重入次数超过最大值
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                // Reentrant acquire
                //写锁可重入一次
                setState(c + acquires);
                return true;
            }
            //c==0还没有人加锁
            //writerShouldBlock--->写锁是否应该阻塞: 这里需要区分公平锁和非公平锁
            //公平锁:如果当前同步队列有老二，那么返回true,否则如果同步队列为空或者当前线程就是老二，返回false
            //非公平锁:总是返回false
            if (writerShouldBlock() ||
            //compareAndSetState--->设置c=1,设置写锁部分为1
                !compareAndSetState(c, c + acquires))
                return false;
                //设置当前线程为锁的拥有者
            setExclusiveOwnerThread(current);
            return true;
        }

2）t2 执行 r.lock，这时进入读锁的 sync.acquireShared(1) 流程，首先会进入 tryAcquireShared 流程。如果有写锁占据，那么 tryAcquireShared 返回 -1 表示失败

tryAcquireShared 返回值表示

-1 表示失败
0 表示成功，但后继节点不会继续唤醒
正数表示成功，而且数值是还有几个后继节点需要唤醒，读写锁返回 1

tryAcquireShared(int unused)源码

protected final int tryAcquireShared(int unused) {
            Thread current = Thread.currentThread();
            int c = getState();
            //查看是否加了写锁，如果!=0表示加了写锁，加了写锁返回true
            if (exclusiveCount(c) != 0 &&
            //如果加了写锁，并且是当前线程自己加的，那么返回false---这里是因为同一个线程可以先加写锁后加读锁
            //完成锁的降级操作
            //如果写锁是另一个线程加的，那么返回false
                getExclusiveOwnerThread() != current)
                //因为这里写锁是t1加的，而t2想要加读锁，因此这里返回-1
                return -1;
            .....          
                return 1;
            }
            return fullTryAcquireShared(current);
        }

3）这时会进入 sync.doAcquireShared(1) 流程，首先也是调用 addWaiter 添加节点，不同之处在于节点被设置为Node.SHARED 模式而非 Node.EXCLUSIVE 模式，注意此时 t2 仍处于活跃状态

4）t2 会看看自己的节点是不是老二，如果是，还会再次调用 tryAcquireShared(1) 来尝试获取锁

5）如果没有成功，在 doAcquireShared 内 for (;;) 循环一次，把前驱节点的 waitStatus 改为 -1，再 for (;😉 循环一次尝试 tryAcquireShared(1) 如果还不成功，那么在 parkAndCheckInterrupt() 处 park

doAcquireShared(int arg)源码

private void doAcquireShared(int arg) {
        //加入节点进队列，如果是一开始没有任何节点在同步队列中，就加入两个节点，一个头结点，一个用来包装线程的节点
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
            //t2线程有没有前驱节点
                final Node p = node.predecessor();
                //如果前驱节点是头节点，说明当前线程还有资格去尝试获取一次锁
                if (p == head) {
                //返回-1表示失败，返回1表示成功
                    int r = tryAcquireShared(arg);
                    //只有获取锁成功后，才能进入下面的逻辑
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                //是否应该在获取失败后阻塞住,将前驱节点的值设置为-1，表示有资格唤醒后继节点
                //然后第一次调用返回false,再次进入循环，尝试获取锁
                if (shouldParkAfterFailedAcquire(p, node) &&
                //第二次来到这里，是阻塞住当前线程,返回的值表示当前线程是否被打断
                    parkAndCheckInterrupt())
                    //如果被打断了，设置打断标记为true
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

t3 r.lock，t4 w.lock

这种状态下，假设又有 t3 加读锁和 t4 加写锁，这期间 t1仍然持有锁，就变成了下面的样子

写锁的解锁流程

t1 w.unlock

这时会走到写锁的 sync.release(1) 流程，调用 sync.tryRelease(1) 成功，变成下面的样子

接下来执行唤醒流程 sync.unparkSuccessor，即让老二恢复运行，这时 t2 在 doAcquireShared 内parkAndCheckInterrupt() 处恢复运行

这回再来一次 for (;😉 执行 tryAcquireShared 成功则让读锁计数加一

这时 t2 已经恢复运行，接下来 t2 调用 setHeadAndPropagate(node, 1)，它原本所在节点被置为头节点

事情还没完，在 setHeadAndPropagate 方法内还会检查下一个节点是否是 shared，如果是则调用doReleaseShared() 将 head 的状态从 -1 改为 0 并唤醒老二，这时 t3 在 doAcquireShared 内parkAndCheckInterrupt() 处恢复运行

这回再来一次 for (;😉 执行 tryAcquireShared 成功则让读锁计数加

下一个节点不是 shared 了，因此不会继续唤醒 t4 所在节点
release(int arg)

public final boolean release(int arg) {
    //尝试释放锁
        if (tryRelease(arg)) {
            Node h = head;
            //头结点是否不为null,并且是否有职责唤醒后继节点
            if (h != null && h.waitStatus != 0)
            //唤醒后继节点
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

tryRelease(int releases)

//释放写锁 
        protected final boolean tryRelease(int releases) {
            //当前线程是否持有独占锁
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            //state进行减的操作，这里是减一
            //这里有锁重入的情况
            int nextc = getState() - releases;
            //判断写锁是否被释放成功
            boolean free = exclusiveCount(nextc) == 0;
            if (free)
            //释放成功了设置当前锁的owner为null
                setExclusiveOwnerThread(null);   
            setState(nextc);
            return free;
        }

被唤醒的线程在doAcquireShared(int arg)方法内被暂停处，恢复运行

private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
            //此时线程2的前驱节点是头结点，因此由资格被优先唤醒
                final Node p = node.predecessor();
                if (p == head) {
                     //尝试去获取锁---这里加锁成功返回1
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                    //断开旧的头结点，将恢复运行的t2节点变成头结点---这个方法是重点，可以看错读锁并发的实现
                        setHeadAndPropagate(node, r);
                        //断开旧的头结点
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                //线程2在此处恢复运行
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

tryAcquireShared(arg);尝试获取读锁成功流程分析

protected final int tryAcquireShared(int unused) {
            Thread current = Thread.currentThread();
             //查看是否加了写锁，如果!=0表示加了写锁，加了写锁返回true
            if (exclusiveCount(c) != 0 &&
            //如果加了写锁，并且是当前线程自己加的，那么返回false---这里是因为同一个线程可以先加写锁后加读锁
            //完成锁的降级操作
            //如果写锁是另一个线程加的，那么返回false
                getExclusiveOwnerThread() != current)
                return -1;
             //成功流程-------------------------------- 
             //获取读锁部分
            int r = sharedCount(c);
            //读锁是否应该被阻塞主
            if (!readerShouldBlock() &&
                 //读锁是否超过最大计数 
                r < MAX_COUNT &&
                //加上65536，即高位加一，表示加了一个读锁
                compareAndSetState(c, c + SHARED_UNIT)) {
                //下面是关于读锁的计数，这里先不进行分析
                if (r == 0) {
                    firstReader = current;
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                    firstReaderHoldCount++;
                } else {
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current))
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                }
                //加锁成功，返回1
                return 1;
            }
            return fullTryAcquireShared(current);
        }

setHeadAndPropagate(Node node, int propagate)

private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        //设置node为头结点
        setHead(node);
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            //获取当前头结点的下一个节点
            Node s = node.next;
            //判断后继节点,如果还是共享锁
            if (s == null || s.isShared())
            //再执行释放锁流程
                doReleaseShared();
        }
    }

doReleaseShared()

private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                //头节点的状态由-1改为0，防止在改动的过程中，其他线程看到头结点的值为-1也进行唤醒操作
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases,失败了重新尝试
                    //唤醒后继节点
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

从setHeadAndPropagate和doReleaseShared()方法可以看出，一旦一个共享节点被唤醒，他就会唤醒他后面连续的所有共享节点，这就是读读并发体现的原理

共享锁的解锁流程

t2 r.unlock，t3 r.unlock

t2 进入 sync.releaseShared(1) 中，调用 tryReleaseShared(1) 让计数减一，但由于计数还不为零

t3 进入 sync.releaseShared(1) 中，调用 tryReleaseShared(1) 让计数减一，这回计数为零了，进入doReleaseShared() 将头节点从 -1 改为 0 并唤醒老二，即

之后 t4 在 acquireQueued 中 parkAndCheckInterrupt 处恢复运行，再次 for (;😉 这次自己是老二，并且没有其他竞争tryAcquire(1) 成功，修改头结点，流程结束

源码分析

写锁上锁流程

static final class NonfairSync extends Sync {
    // ... 省略无关代码
    // 外部类 WriteLock 方法, 方便阅读, 放在此处
    public void lock() {
        sync.acquire(1);
    }
    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final void acquire(int arg) {
        if (
            // 尝试获得写锁失败
                !tryAcquire(arg) &&
                        // 将当前线程关联到一个 Node 对象上, 模式为独占模式
                        // 进入 AQS 队列阻塞
                        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
        ) {
            selfInterrupt();
        }
    }
    // Sync 继承过来的方法, 方便阅读, 放在此处
    protected final boolean tryAcquire(int acquires) {
        // 获得低 16 位, 代表写锁的 state 计数
        Thread current = Thread.currentThread();
        int c = getState();
        int w = exclusiveCount(c);
        if (c != 0) {
            if (
                // c != 0 and w == 0 表示有读锁, 或者
                    w == 0 ||
                            // 如果 exclusiveOwnerThread 不是自己
                            current != getExclusiveOwnerThread()
            ) {
                // 获得锁失败
                return false;
            }
            // 写锁计数超过低 16 位, 报异常
            if (w + exclusiveCount(acquires) > MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            // 写锁重入, 获得锁成功
            setState(c + acquires);
            return true;
        }
        if (
            // 判断写锁是否该阻塞, 或者
                writerShouldBlock() ||
                        // 尝试更改计数失败
                        !compareAndSetState(c, c + acquires)
        ) {
            // 获得锁失败
            return false;
        }
        // 获得锁成功
        setExclusiveOwnerThread(current);
        return true;
    }
    // 非公平锁 writerShouldBlock 总是返回 false, 无需阻塞
    final boolean writerShouldBlock() {
        return false;
    }
}

写锁释放流程

static final class NonfairSync extends Sync {
    // ... 省略无关代码
    // WriteLock 方法, 方便阅读, 放在此处
    public void unlock() {
        sync.release(1);
    }
    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final boolean release(int arg) {
        // 尝试释放写锁成功
        if (tryRelease(arg)) {
            // unpark AQS 中等待的线程
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    // Sync 继承过来的方法, 方便阅读, 放在此处
    protected final boolean tryRelease(int releases) {
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        int nextc = getState() - releases;
        // 因为可重入的原因, 写锁计数为 0, 才算释放成功
        boolean free = exclusiveCount(nextc) == 0;
        if (free) {
            setExclusiveOwnerThread(null);
        }
        setState(nextc);
        return free;
    }
}

读锁上锁流程

package com;
static final class NonfairSync extends Sync {
    // ReadLock 方法, 方便阅读, 放在此处
    public void lock() {
        sync.acquireShared(1);
    }
    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final void acquireShared(int arg) {
        // tryAcquireShared 返回负数, 表示获取读锁失败
        if (tryAcquireShared(arg) < 0) {
            doAcquireShared(arg);
        }
    }
    // Sync 继承过来的方法, 方便阅读, 放在此处
    protected final int tryAcquireShared(int unused) {
        Thread current = Thread.currentThread();
        int c = getState();
        // 如果是其它线程持有写锁, 获取读锁失败
        if (
                exclusiveCount(c) != 0 &&
                        getExclusiveOwnerThread() != current
        ) {
            return -1;
        }
        int r = sharedCount(c);
        if (
            // 读锁不该阻塞(如果老二是写锁，读锁该阻塞), 并且
                !readerShouldBlock() &&
                        // 小于读锁计数, 并且
                        r < MAX_COUNT &&
                        // 尝试增加计数成功
                        compareAndSetState(c, c + SHARED_UNIT)
        ) {
            // ... 省略不重要的代码
            return 1;
        }
        return fullTryAcquireShared(current);
    }
    // 非公平锁 readerShouldBlock 看 AQS 队列中第一个节点是否是写锁
    // true 则该阻塞, false 则不阻塞
    final boolean readerShouldBlock() {
        return apparentlyFirstQueuedIsExclusive();
    }
    // AQS 继承过来的方法, 方便阅读, 放在此处
    // 与 tryAcquireShared 功能类似, 但会不断尝试 for (;;) 获取读锁, 执行过程中无阻塞
    final int fullTryAcquireShared(Thread current) {
        HoldCounter rh = null;
        for (; ; ) {
            int c = getState();
            if (exclusiveCount(c) != 0) {
                if (getExclusiveOwnerThread() != current)
                    return -1;
            } else if (readerShouldBlock()) {
                // ... 省略不重要的代码
            }
            if (sharedCount(c) == MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            if (compareAndSetState(c, c + SHARED_UNIT)) {
                // ... 省略不重要的代码
                return 1;
            }
        }
    }
    // AQS 继承过来的方法, 方便阅读, 放在此处
    private void doAcquireShared(int arg) {
        // 将当前线程关联到一个 Node 对象上, 模式为共享模式
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (; ; ) {
                final Node p = node.predecessor();
                if (p == head) { // 再一次尝试获取读锁
                    int r = tryAcquireShared(arg);
                    // 成功
                    if (r >= 0) {
                        // ㈠
                        // r 表示可用资源数, 在这里总是 1 允许传播
                        //（唤醒 AQS 中下一个 Share 节点）
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (
                    // 是否在获取读锁失败时阻塞（前一个阶段 waitStatus == Node.SIGNAL）
                        shouldParkAfterFailedAcquire(p, node) &&
                                // park 当前线程
                                parkAndCheckInterrupt()
                ) {
                    interrupted = true;
                }
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处
    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        // 设置自己为 head
        setHead(node);
        // propagate 表示有共享资源（例如共享读锁或信号量）
        // 原 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
        // 现在 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            // 如果是最后一个节点或者是等待共享读锁的节点
            if (s == null || s.isShared()) {
                // 进入 ㈡
                doReleaseShared();
            }
        }
    }
    // ㈡ AQS 继承过来的方法, 方便阅读, 放在此处
    private void doReleaseShared() {
        // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
        // 如果 head.waitStatus == 0 ==> Node.PROPAGATE, 为了解决 bug, 见后面分析
        for (; ; ) {
            Node h = head;
            // 队列还有节点
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue; // loop to recheck cases
                    // 下一个节点 unpark 如果成功获取读锁
                    // 并且下下个节点还是 shared, 继续 doReleaseShared
                    unparkSuccessor(h);
                } else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue; // loop on failed CAS
            }
            if (h == head) // loop if head changed
                break;
        }
    }
}
static final class NonfairSync extends Sync {
    // ReadLock 方法, 方便阅读, 放在此处
    public void unlock() {
        sync.releaseShared(1);
    }
    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }
    // Sync 继承过来的方法, 方便阅读, 放在此处
    protected final boolean tryReleaseShared(int unused) {
        // ... 省略不重要的代码
        for (; ; ) {
            int c = getState();
            int nextc = c - SHARED_UNIT;
            if (compareAndSetState(c, nextc)) {
                // 读锁的计数不会影响其它获取读锁线程, 但会影响其它获取写锁线程
                // 计数为 0 才是真正释放
                return nextc == 0;
            }
        }
    }
    // AQS 继承过来的方法, 方便阅读, 放在此处
    private void doReleaseShared() {
        // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
        // 如果 head.waitStatus == 0 ==> Node.PROPAGATE 
        for (; ; ) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                // 如果有其它线程也在释放读锁，那么需要将 waitStatus 先改为 0
                // 防止 unparkSuccessor 被多次执行
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue; // loop to recheck cases
                    unparkSuccessor(h);
                }
                // 如果已经是 0 了，改为 -3，用来解决传播性，见后文信号量 bug 分析
                else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue; // loop on failed CAS
            }
            if (h == head) // loop if head changed
                break;
        }
    }
}

JUC学习之共享模型工具之JUC并发工具包上

AQS 原理

概述

实现不可重入锁

自定义同步器

自定义锁

心得

起源

目标

设计

1) state 设计

2) 阻塞恢复设计

3) 队列设计

ReentrantLock 原理

非公平锁实现原理

加锁流程

解锁流程

加锁流程源码分析:

可重入原理

可打断原理

不可打断模式

可打断模式

公平锁实现原理

条件变量实现原理

await 流程

signal 流程

读写锁原理

ReentrantReadWriteLock

注意事项

缓存

1. 缓存更新策略

2. 读写锁实现一致性缓存

注意

读写锁原理

1. 图解流程

加锁

写锁的解锁流程

共享锁的解锁流程

源码分析

写锁上锁流程

写锁释放流程

读锁上锁流程

相关文章

热门标签

最新文章

目录