理解ReentrantLock

本文的主要内容是理解ReentrantLock源码。

先来看一段代码

public Object take() throws InterruptedException {
	lock.lock();
	try {
		while (count == 0){
			System.out.println("empty,wait with puting....");
			notEmpty.await();
		}
	  // TimeUnit.SECONDS.sleep(1);
		Object x = items[takeptr];
		if (++takeptr == items.length)
			takeptr = 0;
		--count;
		System.out.println(Thread.currentThread().getName()+"----take方法：" +
				"putptr:"+putptr+",takeptr:"+takeptr+",count:"+count);
		/**
		 * 将此条件队列(条件队列的设计是firstWaiter->waiter2->lastWaiter,
		 * 每一个节点都保存一个等待线程)中的第一个等待者转移到同步队列(
		 * 同步队列的格式是head->node->tail,head起标记作用，里面的线程为空
		 * ,tail中的线程不为空)，并且保证同步队列中的第一个节点是可以被唤
		 * 醒的
		 * 
		 */
		notFull.signal();
		return x;
	} finally {
		
		/**
		 * 释放当前锁，并唤醒同步队列中的第一个节点
		 * 
		 */
		lock.unlock();
	}
}

   

先做一个简要的说明，ReentrantLock可以有两个队列存在，一般叫同步队列、条件队列。同步队列就是线程在

lock.lock()的时候阻塞进入的队列，队列的结构：head->node1->node2(tail),head节点表示头节点，起标识

作用不存放线程；条件队列是执行类似notEmpty.await()代码阻塞进入的队列，独立结构:firstWaiter->next

->next2(lastWaiter)，队列中的每一个节点都保存着一个等待线程。

 

上面说的两个队列的结构，现在在来说说他们的状态，先来看看AQS中Node的定义，以下只是一部分

static final class Node {
        /** waitStatus value to indicate thread has cancelled */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
        /** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
        static final Node EXCLUSIVE = null;

       
        volatile int waitStatus;
        volatile Node prev;
        volatile Node next;
        volatile Thread thread;
        Node nextWaiter;

    无论是同步队列还是条件队列，其中的节点类型都是Node，节点的状态有四个值：0,1,-1,-2，后面三个代码中都释，0表示初始值，在同步队列中新增一个节点，它的初始值都是0，同步队列的状态一般是：-1->-1-...->0，就是说只要后面有节点（next不为null），你的状态都要为-1(AQS尽量往这方面做的)，目的就是保证我的前一个节点在释放锁的时候可以唤醒自己；条件队列的状态一般是：-2->-2...->-2。

    节点的状态用waitStatus保存，同步队列中初始为0，在节点后面添加一个节点后，就被修改为-1，条件队列中初始为-2，在把节点从条件队列挪到同步队列时，它的状态会被修改为0，如果此时同步队列中添加一个节点，那么它的状态就被修改为-1，保持这种状态-1->-1..->0;thread用来存放等待线程，prev、next共同构成条件队列，所以它是一个双向队列，nextWaite构成条件队列，它是单向的。

 

 

回到代码中来说

先来看看lock方法

 /**
         * 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);
        }

 

当多个线程竞争锁时，如上代码cas操作成功的将成功获得锁，而竞争失败的将进入同步队列，在这里可以看到这个锁的

实质，它由一个state变量和一个独占线程变量维护（看看他的名字吧，exclusiveOwnerThread），再来看看这个方法可以更加清楚的理解这个锁的实质

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;
        }

 

线程在请求锁的时候都要调用它，上面展示了锁的重入的实现，简单来说就是请求独占锁就是将状态从0原子的修改为1

成功的话把自己标记为独占锁拥有线程，如果重复请求，就把状态累加。

 

如果竞争锁失败，自然就要进入等待队列等待了，看代码

final boolean acquireQueued(final Node node, int arg) {
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } catch (RuntimeException ex) {
            cancelAcquire(node);
            throw ex;
        }
    }

 

     上面代码主要是一个for循环和两个if语句，第一个if语句可以跳出for循环，它想表达的就是等待队列中节点出队的设计

首先条件是，你必须是第一个节点（头结点的下一个节点）并且尝试获得锁成功，这里有点“金蝉脱壳”的味道，头结点

自然会被gc掉，而第一个节点的线程将会跑远，并且把该清空的清空，达到线程出队的效果，第一个节点变为头结点，具体实现在setHead(node)中,第二个if语句有两部分：shouldParkAfterFailedAcquire和parkAndCheckInterrupt，

第一个是用来保证队列的状态-1->-1..->0,第二个是在第一个完成后并尝试获得锁失败后就park当前线程。

 

 

讲完了同步队列的入对，在来看看条件队列，先贴下await的源码

 public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

 

看下英文，就可以了解个大概意思，这就是好

在这里重点来理解下这个while循环，正常情况下线程在条件队列中等待，在被unpark后发现自己在同步队列中了，就跳出循环，以下是isOnSyncQueue(node)的源码

final boolean isOnSyncQueue(Node node) {
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        if (node.next != null) // If has successor, it must be on queue
            return true;
        /*
         * node.prev can be non-null, but not yet on queue because
         * the CAS to place it on queue can fail. So we have to
         * traverse from tail to make sure it actually made it.  It
         * will always be near the tail in calls to this method, and
         * unless the CAS failed (which is unlikely), it will be
         * there, so we hardly ever traverse much.
         */
        return findNodeFromTail(node);
    }

 

节点的状态为CONDITION 当然就是在条件队列中了，node.next不为null自然就在同步队列中了（前面说过的prev和next共同构成同步队列），再没有就通过findNodeFromTail来判断（一般情况是参数node节点是同步队列中的最后一个）。

 

这里有一个很经典的关于中断机制应用的例子，先贴下代码

private int checkInterruptWhileWaiting(Node node) {
            return (Thread.interrupted()) ?
                ((transferAfterCancelledWait(node))? THROW_IE : REINTERRUPT) :
                0;
        }

 

如果线程在park的时候被中断，它将被唤醒，继续执行checkInterruptWhileWaiting方法，首先执行Thread.interrupted(),这个方法会清楚原来的中断状态，在这里如果没有被中断，方法返回0，如果被中断就执行

((transferAfterCancelledWait(node))? THROW_IE : REINTERRUPT)，先解释下THROW_IE ，REINTERRUPT

是什么意思，THROW_IE 的值为-1，表示要抛出中断异常，REINTERRUPT的值为1，表示要再次中断线程，这两个状态

的行为都是因为线程的状态被清除了，这下可以看看transferAfterCancelledWait的代码了

 final boolean transferAfterCancelledWait(Node node) {
        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
            enq(node);
            return true;
        }
        /*
         * If we lost out to a signal(), then we can't proceed
         * until it finishes its enq().  Cancelling during an
         * incomplete transfer is both rare and transient, so just
         * spin.
         */
        while (!isOnSyncQueue(node))
            Thread.yield();
        return false;
    }

 

第一个if表示这个节点还在条件队列中的情况，修改node的状态为0，将节点转移到同步队列中，并且告诉要抛出异常

（return ture），如果此时节点已经在同步队列中了，表示要记得重新中断它。这个时候可以直接到await方法的最后

一个if语句中：如何响应中断，就如字面的意思，为了看看结果可以，可以回到take方法：状态为-1的时候，在take方法中释放了锁，然后由调用take()方法的代码捕获中断异常，状态为1的时候没有抛出异常，但此线程一直被标记为中断

状态，如果程序中有TimeUnit.SECONDS.sleep(1)(会响应中断)的代码，那么他将立即抛出中断异常，线程的中断状态就被清除了，如果没有的话这个异常就被屏蔽掉了，run()方法执行完了，线程也就结束了。

 

接下来简单介绍下signal(),unlock()方法的作用

signal()方法的主要是把条件队列的第一个节点转移到同步队列中(按照同步队列的要求)，并且只能移一个节点；unlock()方法主要是释放锁，并且去唤醒同步队列中的第一个节点。