淮安建设网站制作世界军事新闻
前言
多线程知识中理解了ReentrantLock之后,对于整个AQS也会有大概的理解,后面再去看其它锁的源码就会比较容易。下面带大家一块来学习ReentrantLock源码。
概述
ReentrantLock是可重入的互斥锁,虽然具有与synchronized相同功能,但是会比synchronized更加灵活(具有更多的方法)。ReentrantLock底层基于AbstractQueuedSynchronized。有两种锁方式公平锁和非公平锁,公平锁指线程锁定后会进入队列进行排队等待至获得锁的使用权;而非公平锁指线程锁定后会尝试获取锁,如果获取失败则进入等待队列进行排队。
正文
应用场景
下面有这么一段代码,有10个线程对count进行累加,每个线程累加10000次,那么最终期望的输出值肯定是100000。
private static int count = 0;private static void inrc() {count++;}public static void main(String[] args) throws InterruptedException {for (int i = 0; i < 10; i++) {new Thread(() -> {for (int a=0;a<10000;a++){inrc();}}).start();}TimeUnit.SECONDS.sleep(3);System.out.println(count);}
不加锁的情况,输出会不符合我们的预期。
使用ReentrantLock加锁。
private static Lock lock = new ReentrantLock();private static int count = 0;private static void inrc() {try {//加锁lock.lock();count++;} catch (Exception e) {e.printStackTrace();}finally {lock.unlock();}}public static void main(String[] args) throws InterruptedException {for (int i = 0; i < 10; i++) {new Thread(() -> {for (int a=0;a<10000;a++){inrc();}}).start();}TimeUnit.SECONDS.sleep(3);System.out.println(count);}
输出结果:符合预期
NonfairSync:lock方法
ReentrantLock默认为非公平锁,从其构造方式中可以看出。
public ReentrantLock() {sync = new NonfairSync();}
下面我们来追踪ReentrantLock的lock()方法,由于默认为非公平锁,所以当我们调用lock方法时,实际调用的是NonfairSync的lock方法
public void lock() {sync.lock();}
final void lock() {//尝试通过CAS修改state值,如果此时成功的修改了state的值为1,则证明竞争到了锁资源if (compareAndSetState(0, 1))//将线程占用者属性设置为当前线程setExclusiveOwnerThread(Thread.currentThread());else//如果占用失败,则再次尝试,如果还失败则会加入队列进行等待acquire(1);}
由于是非公平锁,所以进入该方法时,会先尝试获取锁资源,通过CAS(可以保证原子性,即线程安全)设置state值为1,如果成功设置则是拿到了资源,否则调用acquire方法进行尝试。如果是公平锁的话直接调用acquire()方法;
AQS:acquire方法
public final void acquire(int arg) {//tryAcquire(arg):再次尝试获取锁资源//addWaiter(Node.EXCLUSIVE):前面的判断如果获取不到锁资源,则将添加当前线程到队列中进行等待//acquireQueued:自旋尝试获取锁资源//selfInterrupt():如果当前线程的状态被打断了,则将当前线程打断掉(只有当前线程有权限打断自己所在线程)if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}
1、tryAcquire(arg):这里会再次尝试是否能获取到锁资源,有可能当前拥有锁资源的线程就是自己所在的线程,那么这时候是支持锁重入的。
2、acquireQueued(addWaiter(Node.EXCLUSIVE), arg)):添加当前线程到队列中,并以自旋的方式尝试获取锁资源。
3、selfInterrupt():当前线程的状态如果为打断状态,则将当前线程打断掉(只有当前线程有权限打断自己所在线程)
NonfairSync: tryAcquire方法
protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);}
final boolean nonfairTryAcquire(int acquires) {//获取当前线程final Thread current = Thread.currentThread();//获取state的状态看是否被占有int c = getState();//如果为0,则代表当前为被占有if (c == 0) {//通过CAS方式尝试修改其值进行资源占用if (compareAndSetState(0, acquires)) {//成功则将当前线程为资源拥有者setExclusiveOwnerThread(current);return true;}}//判断资源拥有者与当前线程是否是同个,如果是则将state值进行累加,达到可重入的目的else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}
AQS:addWaiter方法
AQS中的队列结构为双向链表结构,每个节点为node对象,该对象拥有前置节点、后置节点属性;
private Node addWaiter(Node mode) {//将当前线程封装成nodeNode node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failure//判断链表的尾部节点是否为空,如果为空则证明整个队列没有值Node pred = tail;if (pred != null) {//设置当前节点的前置节点node.prev = pred;//通过CAS的方式将当前节点与上个节点关联起来。通过CAS保证线程安全if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}enq(node);return node;}
AQS: acquireQueued方法
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 GCfailed = false;return interrupted;}//判断当前线程的状态是否为等待中,如果是则返回true,不是则将状态设置为等待状态if (shouldParkAfterFailedAcquire(p, node) &&//阻塞并返回当前线程是否被打断标识parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}
AQS:shouldParkAfterFailedAcquire方法
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;//如果上个节点处于等待状态,则代表当前的线程可以进行阻塞,因为上个线程都没获取资源,肯定不会轮到自己的,可以安心阻塞if (ws == Node.SIGNAL)/** This node has already set status asking a release* to signal it, so it can safely park.*/return true;//状态大于0,证明上个线程取消了,需要从队列中往前查一个未取消的if (ws > 0) {/** Predecessor was cancelled. Skip over predecessors and* indicate retry.*/do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {/** waitStatus must be 0 or PROPAGATE. Indicate that we* need a signal, but don't park yet. Caller will need to* retry to make sure it cannot acquire before parking.*///将当前节点设置为等待状态compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}
AQS:parkAndCheckInterrupt
private final boolean parkAndCheckInterrupt() {LockSupport.park(this);return Thread.interrupted();}
public static void park(Object blocker) {Thread t = Thread.currentThread();setBlocker(t, blocker);//调用park方法进行阻塞,这样调用的是native方法,即调用内核态库中的函数UNSAFE.park(false, 0L);setBlocker(t, null);}
AQS:cancelAcquire方法
private void cancelAcquire(Node node) {// Ignore if node doesn't existif (node == null)return;//将当前线程置为null,方便gcnode.thread = null;// Skip cancelled predecessors//获取上个节点Node pred = node.prev;while (pred.waitStatus > 0)node.prev = pred = pred.prev;// predNext is the apparent node to unsplice. CASes below will// fail if not, in which case, we lost race vs another cancel// or signal, so no further action is necessary.Node predNext = pred.next;// Can use unconditional write instead of CAS here.// After this atomic step, other Nodes can skip past us.// Before, we are free of interference from other threads.//将线程状态设置为取消node.waitStatus = Node.CANCELLED;// If we are the tail, remove ourselves.//如果当前线程位于队列的最后一个位置,由于取消了,所以要将上个队列位置的线程置为尾部if (node == tail && compareAndSetTail(node, pred)) {compareAndSetNext(pred, predNext, null);} else {// If successor needs signal, try to set pred's next-link// so it will get one. Otherwise wake it up to propagate.int ws;//这一步只要将上个线程与下个线程进行关联,移除当前线程在队列的位置if (pred != head &&((ws = pred.waitStatus) == Node.SIGNAL ||(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&pred.thread != null) {Node next = node.next;if (next != null && next.waitStatus <= 0)compareAndSetNext(pred, predNext, next);} else {//释放下一个线程阻塞状态unparkSuccessor(node);}node.next = node; // help GC}}
AQS:unparkSuccessor
private void unparkSuccessor(Node node) {/** If status is negative (i.e., possibly needing signal) try* to clear in anticipation of signalling. It is OK if this* fails or if status is changed by waiting thread.*/int ws = node.waitStatus;if (ws < 0)compareAndSetWaitStatus(node, ws, 0);/** Thread to unpark is held in successor, which is normally* just the next node. But if cancelled or apparently null,* traverse backwards from tail to find the actual* non-cancelled successor.*/Node s = node.next;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);}
NonfairSync:unlock方法
通过前面的步骤,我们知道处于队列中的线程都处于阻塞状态,持有资源的线程执行完之后,需要调用unlock方法将队列中的下一个线程唤醒,并将自己从队列中进行移除。
public void unlock() {sync.release(1);}
AQS:release方法
public final boolean release(int arg) {//判断是否能释放,因为ReentrantLock是重入锁,每次重入时state会加1,所以如果state-1不等于0的话,需要多次的unlock才能达到释放资源。比如同个线程中调用了两次lock,需要调用两次unlock才能释放资源。if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)//将下个节点唤醒unparkSuccessor(h);return true;}return false;}
ReentrantLock:tryRelease方法
protected final boolean tryRelease(int releases) {//将state减一int c = getState() - releases;if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free = false;//如果为0释放资源if (c == 0) {free = true;setExclusiveOwnerThread(null);}//不为0,证明之前有过重入的情况,需要等该线程的所有unlock方法执行完毕后,state等于0才能释放setState(c);return free;}
AQS:unparkSuccessor方法
private void unparkSuccessor(Node node) {/** If status is negative (i.e., possibly needing signal) try* to clear in anticipation of signalling. It is OK if this* fails or if status is changed by waiting thread.*/int ws = node.waitStatus;//将当前线程状态设置为0if (ws < 0)compareAndSetWaitStatus(node, ws, 0);/** Thread to unpark is held in successor, which is normally* just the next node. But if cancelled or apparently null,* traverse backwards from tail to find the actual* non-cancelled successor.*///查找下个节点,状态不为取消的Node s = node.next;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);}
总结
下面画了流程图,帮助大家更好的理解。