ReentrantLock源码解析(上)

源码解析(上)"/>

ReentrantLock源码解析(上)

序 -- 早年间由于synchronized关键字的效率问题,导致在jdk1.6版本之前,著名的daog lee先生编写出了ReentrantLock提供给我们使用,一方面是因为synchronized锁在jdk1.6之间是一把重量级锁,无论是线程的交替执行,或者是并发执行,都会调用OS系统的函数,导致及其耗费时间,ReentrantLock却在java的层面上大幅度的解决了这一问题,今天,我们就来研究一下ReentrantLock的源码,看看它是如何做到的性能优化,

正文 -- ReentrantLock的公平锁(锁定机制)的实现

    ReentrantLock锁的实现其实并不困难,我们可以先大致浏览一下它的实现过程:

volatile int status=0;
Queue parkQueue;//集合 数组  listvoid lock(){while(!compareAndSet(0,1)){ CAS操作//park();----将线程休眠,以保证cpu不背过高占用}}void unlock(){lock_notify();
}void park(){//将当期线程加入到等待队列parkQueue.add(currentThread);//将当期线程释放cpu  阻塞   睡眠releaseCpu(); 
}
void lock_notify(){//status=0//得到要唤醒的线程头部线程Thread t=parkQueue.header();//唤醒等待线程unpark(t);
}

上面我们利用了伪代码的方式,将ReentrantLock的实现简化.如果大家对上述的过程有锁疑问,建议先学习多线程基本再来阅读本文.

在基本了解了如何大致实现一把锁的过程后,我们便可以开始着手研究ReentrantLock的源码了.

第一个是ReentrantLock的lock()方法,也就是上锁的方法,在调用ReentrantLock对象进行上锁时,ReentrantLock内部有一个AQS即双向队列同步器,简而言之就是一个需要线程排队的队列,该队列由一条带头尾节点的双向链表进行维护.如果在当前队列中没有线程正在排队,即可以直接获取锁,当前有线程正在排队,则入队等待,并且休眠.

AQS队列中的每一个节点均为一个Node对象,他拥有前节点,以及后节点,该Node节点含有的线程,以及一个waitState状态码:

static final class Node {volatile AbstractQueuedSynchronizer.Node prev;volatile AbstractQueuedSynchronizer.Node next;volatile int waitStatus;//剩余的参数省略
}

public class ReentrantLock implements Lock, Serializable {//首先进入lock方法,方法中调用了AQS队列中的acquire方法,其中含有一个默认的参数1,代表该线程期望获取这把锁.public void lock() {this.sync.acquire(1);}//在调用acquire方法以后,正式进行公平锁的实现过程,其中的逻辑条件非常简单//第一,if条件中有两大函数,tryAcquire方法代表获取锁的过程,失败返回false,成功返回true//成功以后代码开始返回,即已经获取锁开始执行线程中的方法,失败则调用acquireQueued方法进行排队//acquireQueued方法中调用了addWaiter方法,即将这个线程生成为Node对象,然后加入AQS队列中等待.public final void acquire(int arg) {if (!this.tryAcquire(arg) && this.acquireQueued(this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE), arg)) {selfInterrupt();}}
}

紧接着上述代码,首先我们进去tryAcquire方法中获取锁:

static final class FairSync extends ReentrantLock.Sync{//公平锁的实现类,其中其他方法被省略.@ReservedStackAccessprotected final boolean tryAcquire(int acquires) {//首先获取当前线程Thread current = Thread.currentThread();//获取此时锁的状态码,即时候有人占用锁,没有人返回0,有人返回1.int c = this.getState();//当锁的状态码为0时,分为两种情况.if (c == 0) {//首先进入hasQueuedPredecessors方法中,判断当前队列是否有线程正在排队.//当有线程正在排队时,代表锁还在被释放的过程中,仅仅只是锁的状态码被改变,但锁没有被释放,所以线程仍需入队排队.//当进入hasQueuedPredecessors方法中,发现没有线程正在排队时,代表锁此时并没有被人持有,此时可以尝试CAS操作获取锁,当获取成功,tryAcquire方法返回true,线程继续执行.if (!this.hasQueuedPredecessors() && this.compareAndSetState(0,acquires)) {this.setExclusiveOwnerThread(current);return true;}//else if中会判断是否当前线程正在持有锁,即锁的重入机制} else if (current == this.getExclusiveOwnerThread()) {//当当前线程持有锁时,nextc会自增,在当前线程执行完时,会自减保证重入机制的顺利进行.int nextc = c + acquires;if (nextc < 0) {throw new Error("Maximum lock count exceeded");}//ReentrantLock对象本省的state状态码即表示重入锁的进行进度.this.setState(nextc);return true;}return false;}
}

接下来我们看看hasQueuedPredecessors方法中判断队列为空的具体实现步骤:

public final boolean hasQueuedPredecessors() {AbstractQueuedSynchronizer.Node h;//首先获取AQS队列中的头结点,判断是否为空,如果为空,则代表队列为空,直接返回false,代表可以尝试CAS操作获取锁.if ((h = this.head) != null) {//如果判断队头不为空,代表已经有人开始排队.AbstractQueuedSynchronizer.Node s;//获取第一个排队的人,以及排队的人的waitState状态码.if ((s = h.next) == null || s.waitStatus > 0) {s = null;for(AbstractQueuedSynchronizer.Node p = this.tail; p != h && p != null; p = p.prev) {if (p.waitStatus <= 0) {s = p;}}}//如果发现排队的人不为null,并且排队的线程不是当前线程,则代表需要排队,直接返回true,跳过CAS获取锁的操作.if (s != null && s.thread != Thread.currentThread()) {return true;}}return false;}

此时最开始acquire方法中的if两大判断条件已经完成了第一条,成功获取锁后,方法直接返回,没有成功获取锁,则进入if的第二大判断条件:this.acquireQueued(this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE), arg))

在这里有个要注意的点是,addWaiter方法在网上大部分流传着一个旧版本,目前jdk11已经加以优化为新版本,但是意思大同小异,旧版本便于阅读,我们以旧版本进行分析:

private Node addWaiter(Node mode) { //Node.EXCLUSIVE = null//将当前线程封装成Node，并且mode为独占锁Node node = new Node(Thread.currentThread(), mode); // tail是AQS的中表示同步队列队尾的属性，刚开始为null，所以进行enq(node)方法Node pred = tail;if (pred != null) { //tail不为空的情况，说明队列中存在节点数据node.prev = pred;  //将当前线程的Node的prev节点指向tailif (compareAndSetTail(pred, node)) {//通过cas讲node添加到AQS队列pred.next = node;//cas成功，把旧的tail的next指针指向新的tailreturn node;}}enq(node); //tail=null，将node添加到AQS队列中return node;}

private Node enq(final Node node) {//自旋for (;;) {Node t = tail; //如果是第一次添加到队列，那么tail=nullif (t == null) { // Must initialize//CAS的方式创建一个空的Node作为头结点if (compareAndSetHead(new Node()))//此时队列中只一个头结点，所以tail也指向它tail = head;} else {//进行第二次循环时，tail不为null，进入else区域。将当前线程的Node结点的prev指向tail，然后使用CAS将tail指向Nodenode.prev = t;if (compareAndSetTail(t, node)) {//t此时指向tail,所以可以CAS成功，将tail重新指向Node。此时t为更新前的tail的值，即指向空的头结点，t.next=node，就将头结点的后续结点指向Node，返回头结点t.next = node;return t;}}}}总体即维护链表,将节点入队,并设置pre与next指针.

至此我们的链表已经基本维护完毕,下一步也是最为关键的一步,即acquireQueued方法,该方法的参数便是在上图中创建的node节点.

final boolean acquireQueued(AbstractQueuedSynchronizer.Node node, int arg) {//判断是否需要打断该线程的标机,暂时不需要管.boolean interrupted = false;try {//首先进入死循环,为我们判断该线程是否休眠做准备,while(true) {//首先获取当前线程的上一个节点AbstractQueuedSynchronizer.Node p = node.predecessor();//如果发现上一个节点是队头,那么我们需要再去尝试获取一次锁//因为在并发编程中,可能存在上一个节点在最开始的lock获取锁时没有获取到,但是在接下来代码执行过程中,前一个节点释放了锁,如果在这里再获取一次,就可以减少程序的花销.if (p == this.head && this.tryAcquire(arg)) {this.setHead(node);p.next = null;return interrupted;}//当程序没有获取到锁或者程序的上一个节点并不是队头,那我们我们就需要考虑这个线程是否应该被休眠,即park//于是我们进入shouldParkAfterFailedAcquire方法进行判断//这个方法的两个参数分别是当前线程的上一个节点,以及当前线程自己的节点.if (shouldParkAfterFailedAcquire(p, node)) {interrupted |= this.parkAndCheckInterrupt();}}} catch (Throwable var5) {this.cancelAcquire(node);if (interrupted) {selfInterrupt();}throw var5;}}

来,接着我们进入到shouldParkAfterFailedAcquire方法中:

private static boolean shouldParkAfterFailedAcquire(AbstractQueuedSynchronizer.Node pred, AbstractQueuedSynchronizer.Node node) {//首先获取上一个节点的waitStatus,即等待情况//当上一个节点ws为-1时,即上一个节点正在休眠时,直接返回,并将当前线程进入休眠状态.//在这里需要注意的是,当前线程的休眠状态ws只能由下一个节点帮忙修改,具体原因会在下面的代码中说明.int ws = pred.waitStatus;if (ws == -1) {return true;} else {//当ws大于0时,代表线程等待超市,需要被中断.if (ws > 0) {do {node.prev = pred = pred.prev;} while(pred.waitStatus > 0);pred.next = node;//当线程等于0时,我们会将上一个线程修改为休眠状态,并再次自旋进行判断//而第二次进入该方法时,由于第一次的修改,我们已经将上一个线程的ws改为了-1,则直接休眠当前线程} else {pred.compareAndSetWaitStatus(ws, -1);}return false;}}

最后在万事俱备之时,我们终于走到了线程休眠的这一步,也就是parkAndCheckInterrupt方法:

方法很短,调用了UnSafe类中的park方法,将这个线程在该处休眠.但是同时我们在上个代码段中提到的问题在这里就可以解决了,为什么?我们需要在当前线程中,修改上一个线程的ws休眠状态,而不是修改我们自己的休眠状态.

private final boolean parkAndCheckInterrupt() {LockSupport.park(this);//线程将在此处休眠,当被唤醒时,线程会直接被acquireQueued方法返回,并完成任务.return Thread.interrupted();}

其实当前线程做不到修改自己的ws休眠状态,如果说在休眠之前将状态进行修改,那么如果在park方法将改线程休眠出错时,那么就与ws的状态是互相矛盾的,如果将ws状态的修改放在park之后就更加不可能了,因为当线程能执行park以后的代码时,就已经退出休眠状态了.

好了,关于ReentrantLock公平锁的锁定源码分析到这里也就结束了,下一章我们将进行ReentrantLock解锁源码的分析.