从ReentrantLock到AQS的分析
ReentrantLock的主要方法:
1、lock():获取锁,没获取到会阻塞(Sync定义的抽象方法)
2、unLock():释放锁(ReentrantLock定义的方法)
3、tryLock():内部调用nonfairTryAcquire(1)以非公平的方式尝试获取锁,不会阻塞
4、tryLock(long timeout, TimeUnit unit):调用各自重写的tryAcquire()方法来尝试获取锁,也会到队列排队,超过等待时间没获取到会中断等待
5、newCondition():设置条件锁,唤醒的时候可以根据条件来唤醒对应的锁
6、lockInterruptibly():可中断锁;持有该锁的线程或者在等待该锁的线程可以被中断
ReentrantLock可以实现公共锁和非公平锁,内部类Sync继承AQS来实现主要功能;
AQS主要是通过CAS+volatile int state标识+双向链表构成的先进先出队列实现获取锁、释放锁和等待锁的过程,ReentrantLock也遵循这一核心;
ReentrantLock的非公平锁:
非公平锁的获取:state=0(没有线程拥有锁),这时先尝试获取锁,没有获取到的话再进入队列排队
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
// 默认为非公平锁 public ReentrantLock() { sync = new ReentrantLock.NonfairSync(); } static final class NonfairSync extends ReentrantLock.Sync { /** * Performs lock. Try immediate barge, backing up to normal * acquire on failure. */ final void lock() { // 与公平锁的不同之处,会先试着抢占锁,没抢到再排队等待 if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else // 进入队列排队,等待获得锁 acquire(1); } // 注意:子类重写了AQS的tryAcquire()方法,子类调用是调用自己的 // 公平锁和非公平锁的tryAcquire()不一样 protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } } // 非公平锁的tryAcquire()方法 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()) { // 可重入 int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; }
ReentrantLock的公平锁:
公平锁获取:state=0(没有线程拥有锁),这时判断队列是否有线程在等待,如果有的话则直接进入队列排队
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
// 设置是公平锁还是非公平锁 public ReentrantLock(boolean fair) { sync = fair ? new ReentrantLock.FairSync() : new ReentrantLock.NonfairSync(); } static final class FairSync extends ReentrantLock.Sync { final void lock() { // 这里不会像非公平锁一样先尝试获取锁,而是队列中如果有等待线程,则直接进入队列排队 acquire(1); } // 公平锁的tryAcquire() protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { // 公平锁在state=0(没有线程持有锁)的时候,会先去判断等待队列里面是否有线程再排队,使用hasQueuedPredecessors()来判断 // 如果有线程在排队则不能获取到锁 // compareAndSetState(0, acquires) CSA原子更新 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"); // 上面的compareAndSetState和setState的区别: // 上面线程还没有获得锁,无法保证线程安全,所以会调用unsafe.compareAndSwapInt()来保证state更新的原子性 // 这里的线程已经获的了锁,是线程安全的,可以直接修改state的值 setState(nextc); return true; } return false; } }
锁的释放:Sync实现了AQS的tryRelease()方法来释放锁,释放的时候会唤醒后继等待节点
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
public void unlock() { sync.release(1); } // AQS中的release方法,释放锁 public final boolean release(int arg) { // 调用Sync的tryRelease() if (tryRelease(arg)) { Node h = head; // 如果h.waitStatus!=0,说明有后继节点; // 因为入队列的节点的ws默认初始化为0,如果唤醒时ws=0,说明没有节点更新前驱节点的ws,这说明没有线程入队列 if (h != null && h.waitStatus != 0) // 释放锁的同时,唤醒队列中最前面的waitStatus<=0的等待线程 unparkSuccessor(h); return true; } return false; } // Sync实现tryRelease() protected final boolean tryRelease(int releases) { int c = getState() - releases; // 如果当前线程不是持有锁的线程则会报错,例如在线程执行的过程中把锁给换了 if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; if (c == 0) { free = true; setExclusiveOwnerThread(null); } setState(c); return free; } // 释放锁的同时,唤醒后继节点 private void unparkSuccessor(Node node) { int ws = node.waitStatus; // 为什么要设置头节点的ws为0 if (ws < 0) compareAndSetWaitStatus(node, ws, 0); // 头节点的下一个节点,如果不是null并且ws<=0则可以直接唤醒 Node s = node.next; // 后继节点为什么会等于null? 插入节点的时候pred.next不是线程安全的,所以有可能为null,但是从尾部遍历一定是可以的 if (s == null || s.waitStatus > 0) { s = null; for (Node t = tail; t != null && t != node; t = t.prev) // 找到队列中第一个waitStatus<=0的节点,并唤醒;后继节点为什么会出现=0的状态? 新加入的tail节点,还没更新状态 if (t.waitStatus <= 0) s = t; } if (s != null) // 唤醒在自旋过程中被挂起的节点,继续自旋获取锁 LockSupport.unpark(s.thread); }
AbstractQueuedSynchronizer:
AQS遵循模板方法的设计模式:定义好了算法的骨架(acquire(1)和release(1)),将某些步骤推迟到子类中实现(tryAcquire()和tryRelease());骨架的逻辑是固定的,但里面的一些方法可以由子类来实现;
AQS的两种模式:
独占模式(一次只能有一个线程访问资源):ReentrantLock使用的是独占模式
共享模式(一次可以有多个线程访问资源):Semaphore使用的是共享模式;
在上面的锁的获取过程中,无论是非公平锁还是公平锁,线程如果没有得到锁,会调用acquire(1)方法进入队列排队等待,队列主要是在AQS里面实现的,下面介绍线程如何进入队列排队等待:
AQS的acquire(1):没有获取到锁的线程就进队列
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
// AQS的acquire() public final void acquire(int arg) { // 先调用自己的tryAcquire()尝试获取锁,成功获取的话则返回true,该方法结束 // 没获取到的话则执行后面的入队操作,到队列中等待锁的获取 if (!tryAcquire(arg) && acquireQueued(addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE), arg)) selfInterrupt(); }
线程进入队列:acquireQueued(addWaiter(Node.EXCLUSIVE), arg),抽象类实现了这个方法,并且子类不能重写;
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
// ReentrantLock为独占模式 private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); Node pred = tail; // 如果队列中有线程,则尝试将当前线程加到队尾 if (pred != null) { node.prev = pred; // 通过CAS原子更新将当前线程插入到队尾(如果有线程更新了尾结点,则插入失败,调用enq()来插入) if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } // 队列为空或者原子更新失败 enq(node); return node; } private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { // Must initialize // 此时队列为空,初始化队列;队列通过双向链表来实现,head节点没有线程 if (compareAndSetHead(new Node())) tail = head; } else { // 此时队列不为空,通过CAS将线程插入队尾,如果插入成功则返回节点,失败则进入下一轮循环 // 这里没有保证线程安全,可能出现尾分叉 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
// 队列中的线程都通过自旋来获取锁 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; } // 没拿到锁,判断是否将当前节点挂起 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; // 如果节点的前驱节点的ws==SIGNAL,则返回true,将当前节点挂起,不用再一直尝试获取锁; // 在前驱节点释放之后,这个节点会被唤醒,唤醒之后再自旋获取锁 if (ws == Node.SIGNAL) /* * This node has already set status asking a release to signal it, so it can safely park. */ return true; // ws>0表示该节点的线程取消排队,将该节点踢出队列 if (ws > 0) { do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { /* * waitStatus must be 0 or PROPAGATE. */ // 此时前驱节点的ws不为SIGNAL,本轮循环将前驱节点的ws设置为SIGNAL,在下一轮循环,可以将当前节点挂起 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }
state安全更新的重要保证:volatile和CAS
state是线程是否持有锁的一个标志,安全的修改state是锁的线程安全的基本保障,volatile可以保证state更新后对其他的线程可见,但是不能保证更新的原子性,所以每次使用CAS来更新state保证原子性;节点入队也是CAS更新
CAS(比较并交换):每次更新变量的时候,比较变量在内存中的值和所给的旧期望值是不是一致;
1、如果一致则说明没有其他线程修改变量,这时候可以将变量设置为新的值;
2、如果不一致则说明变量被其他的线程修改了,这时候通过自旋来等待下一次修改,直到修改成功 或者 直接修改失败;
采取乐观锁的思想:认为访问数据时不会产生并发冲突,但是在更新变量时会检查变量是否被其他的线程修改过;
优点:不需要加锁,没有上下文切换带来的开销
存在的问题:
1、长时间的自旋消耗CPU;
2、ABA问题,操作过程中A变量有可能被别的线程修改过,可以通过在变量前加一个版本号来解决,每次不仅对比变量值,还对比版本号;
3、CAS只能保证一个变量的原子操作;