AbstractQueuedSynchronizer 기반 동시성 클래스에 대한 자세한 설명
- 零下一度원래의
- 2017-07-17 14:58:471352검색
Fair 모드 ReentrantLock 구현 원리
이전 기사에서는 AbstractQueuedSynchronizer의 독점 잠금과 공유 잠금을 연구했는데, 처음 두 기사의 기초를 바탕으로 승리를 활용하고 동시성이 어떻게 이루어지는지 살펴볼 수 있습니다. AbstractQueuedSynchronizer 기반 클래스가 구현되었습니다.
ReentrantLock은 분명히 배타적 잠금입니다, 우선 공정 모드의 ReentrantLock, Sync는 ReentractLock의 기본 클래스이며 AbstractQueuedSynchronizer에서 상속됩니다. 코드 구현을 살펴보세요.
1 abstract static class Sync extends AbstractQueuedSynchronizer { 2 private static final long serialVersionUID = -5179523762034025860L; 3 4 /** 5 * Performs {@link Lock#lock}. The main reason for subclassing 6 * is to allow fast path for nonfair version. 7 */ 8 abstract void lock(); 9 10 /**11 * Performs non-fair tryLock. tryAcquire is12 * implemented in subclasses, but both need nonfair13 * try for trylock method.14 */15 final boolean nonfairTryAcquire(int acquires) {16 final Thread current = Thread.currentThread();17 int c = getState();18 if (c == 0) {19 if (compareAndSetState(0, acquires)) {20 setExclusiveOwnerThread(current);21 return true;22 }23 }24 else if (current == getExclusiveOwnerThread()) {25 int nextc = c + acquires;26 if (nextc < 0) // overflow27 throw new Error("Maximum lock count exceeded");28 setState(nextc);29 return true;30 }31 return false;32 }33 34 protected final boolean tryRelease(int releases) {35 int c = getState() - releases;36 if (Thread.currentThread() != getExclusiveOwnerThread())37 throw new IllegalMonitorStateException();38 boolean free = false;39 if (c == 0) {40 free = true;41 setExclusiveOwnerThread(null);42 }43 setState(c);44 return free;45 }46 47 protected final boolean isHeldExclusively() {48 // While we must in general read state before owner,49 // we don't need to do so to check if current thread is owner50 return getExclusiveOwnerThread() == Thread.currentThread();51 }52 53 final ConditionObject newCondition() {54 return new ConditionObject();55 }56 57 // Methods relayed from outer class58 59 final Thread getOwner() {60 return getState() == 0 ? null : getExclusiveOwnerThread();61 }62 63 final int getHoldCount() {64 return isHeldExclusively() ? getState() : 0;65 }66 67 final boolean isLocked() {68 return getState() != 0;69 }70 71 /**72 * Reconstitutes this lock instance from a stream.73 * @param s the stream74 */75 private void readObject(java.io.ObjectInputStream s)76 throws java.io.IOException, ClassNotFoundException {77 s.defaultReadObject();78 setState(0); // reset to unlocked state79 }80 }Sync는 Public 클래스에 속하며, Sync가 상속된다는 추상적인 설명입니다. Sync가 주로 수행하는 작업을 간략하게 요약해 보겠습니다(Sync는 ReentrantLock의 공정한 잠금의 핵심이 아니기 때문입니다).
는 잠금 방법을 정의합니다. 일반적으로 ReentrantLock의 lock() 메소드를 호출할 수 있는 이유는 Sync가 이를 정의하기 때문입니다.
불공평한 잠금 tryAcquira 메소드를 구현합니다.
tryRelease 메소드를 구현합니다. 단순, 상태 -1, 배타적 잠금 스레드가 비어 있습니다.
isHeldExclusively 메소드를 구현합니다.
은 newCondition 메소드를 정의하여 개발자가 Condition을 사용하여 알림/대기를 구현할 수 있도록 합니다.
다음으로 공정한 잠금을 살펴보겠습니다. Sync를 상속한 FairSync 클래스의 구현:
1 static final class FairSync extends Sync { 2 private static final long serialVersionUID = -3000897897090466540L; 3 4 final void lock() { 5 acquire(1); 6 } 7 8 /** 9 * Fair version of tryAcquire. Don't grant access unless10 * recursive call or no waiters or is first.11 */12 protected final boolean tryAcquire(int acquires) {13 final Thread current = Thread.currentThread();14 int c = getState();15 if (c == 0) {16 if (!hasQueuedPredecessors() &&17 compareAndSetState(0, acquires)) {18 setExclusiveOwnerThread(current);19 return true;20 }21 }22 else if (current == getExclusiveOwnerThread()) {23 int nextc = c + acquires;24 if (nextc < 0)25 throw new Error("Maximum lock count exceeded");26 setState(nextc);27 return true;28 }29 return false;30 }31 }핵심 사항을 정리해 보겠습니다.
획득할 때마다 state+1, 현재 스레드가 lock()이고 그 다음 lock()이면 상태는 +1을 유지하고 Unlock()일 때 해당 상태는 -1입니다. 상태가 0으로 줄어들 때까지 이는 현재 스레드가 모든 상태를 해제했음을 의미합니다. 다른 스레드는
state=0 시간을 경쟁할 수 있으며, hasQueuedPredecessors 메서드를 통해 판단하고, hasQueuedPredecessors의 구현은 "h != t && ((s = h.next) == null || s.thread != Thread.currentThread()); ", 여기서 h는 head이고 t는 tail입니다. 코드에서 결과가 반전되므로 반전 후의 판단은 "h == t | | ((s = h.next) != null && s.thread == Thread.currentThread());", 요약하면 !hasQueuedPredecessors()로 판단할 수 있는 상황은 두 가지입니다.
-
h==t, h==t의 상황은 또는 현재 FIFO 대기열에 데이터가 없습니다,또는 하나의 헤드만 구축되었으며 어떤 노드에도 연결되지 않았습니다 , 그러니까 머리가 꼬리
(s = h.next)! = null && s.thread == Thread.currentThread(), 현재 스레드는 첫 번째 노드에 있는 스레드입니다. waiting
현재 스레드보다 오래 기다리는 스레드가 없으면 가서 획득 작업을 실행한 다음 CAS 작업을 통해 tryAcquire가 0에서 1까지 성공하는 스레드
스레드 tryAcquire가 성공하지 못한 경우 tryAcquire 순서대로 FIFO 큐로 구성됩니다. 즉 첫 번째 tryAcquire가 실패하면 선두 다음으로 1순위, 두 번째 tryAcquire가 실패하면 선두 다음으로 2순위
때 tryAcquire에 성공한 스레드가 릴리스되고, tryAcquire에 실패한 첫 번째 스레드가 가장 먼저 tryAcquire를 시도합니다. 이는 선착순, 일반적인 공정한 잠금입니다
Unfair 모드 ReentrantLock 구현 원칙
공정 모드 ReentrantLock을 읽은 후 불공정 모드 ReentrantLock이 어떻게 구현되는지 살펴보겠습니다. Sync 클래스에서도 상속되는 NonfairSync 클래스는 다음과 같이 구현됩니다.
1 static final class NonfairSync extends Sync { 2 private static final long serialVersionUID = 7316153563782823691L; 3 4 /** 5 * Performs lock. Try immediate barge, backing up to normal 6 * acquire on failure. 7 */ 8 final void lock() { 9 if (compareAndSetState(0, 1))10 setExclusiveOwnerThread(Thread.currentThread());11 else12 acquire(1);13 }14 15 protected final boolean tryAcquire(int acquires) {16 return nonfairTryAcquire(acquires);17 }18 }结合nonfairTryAcquire方法一起讲解，nonfairTryAcquire方法的实现为：
1 final boolean nonfairTryAcquire(int acquires) { 2 final Thread current = Thread.currentThread(); 3 int c = getState(); 4 if (c == 0) { 5 if (compareAndSetState(0, acquires)) { 6 setExclusiveOwnerThread(current); 7 return true; 8 } 9 }10 else if (current == getExclusiveOwnerThread()) {11 int nextc = c + acquires;12 if (nextc < 0) // overflow13 throw new Error("Maximum lock count exceeded");14 setState(nextc);15 return true;16 }17 return false;18 }看到差别就在于非公平锁lock()的时候会先尝试通过CAS看看能不能把state从0变为1（即获取锁），如果可以的话，直接获取锁而不需要排队。举个实际例子就很好理解了：
线程1、线程2、线程3竞争锁，线程1竞争成功获取锁，线程2、线程3依次排队
线程1执行完毕，释放锁，state变为0，唤醒了第一个排队的线程2
此时线程4来尝试获取锁了，由于线程2被唤醒了，因此线程2与线程4竞争锁
线程4成功将state从0变为1，线程2竞争锁失败，继续park
看到整个过程中，后来的线程4反而比先来的线程2先获取锁，相当于是一种非公平的模式，
那为什么非公平锁效率会比公平锁效率高？上面第（3）步如果线程2和线程4不竞争锁就是答案。为什么这么说，后面的解释很重要，希望大家可以理解：
<span style="color: #000000">线程1是先将state设为0，再去唤醒线程2，这两个过程之间是有时间差的。<br/><br/>那么如果线程1将state设置为0的时候，线程4就通过CAS算法获取到了锁，且在线程1唤醒线程2之前就已经使用完毕锁，那么相当于线程2获取锁的时间并没有推迟，在线程1将state设置为0到线程1唤醒线程2的这段时间里，反而有线程4获取了锁执行了任务，这就增加了系统的吞吐量，相当于单位时间处理了更多的任务。</span>
从这段解释我们也应该能看出来了，非公平锁比较适合加锁时间比较短的任务。这是因为加锁时间长，相当于线程2将state设为0并去唤醒线程2的这段时间，线程4无法完成释放锁，那么线程2被唤醒由于没法获取到锁，又被阻塞了，这种唤醒-阻塞的操作会引起线程的上下文切换，继而影响系统的性能。
Semaphore实现原理
Semaphore即信号量，用于控制代码块的并发数，将Semaphore的permits设置为1相当于就是synchronized或者ReentrantLock，Semaphore具体用法可见Java多线程19：多线程下的其他组件之CountDownLatch、Semaphore、Exchanger。信号量允许多条线程获取锁，显然它的锁是一种共享锁，信号量也有公平模式与非公平模式，相信看懂了上面ReentrantLock的公平模式与非公平模式的朋友应该对Semaphore的公平模式与非公平模式理解起来会更快，这里就放在一起写了。
首先还是看一下Semaphore的基础设施，它和ReentrantLock一样，也有一个Sync：
1 abstract static class Sync extends AbstractQueuedSynchronizer { 2 private static final long serialVersionUID = 1192457210091910933L; 3 4 Sync(int permits) { 5 setState(permits); 6 } 7 8 final int getPermits() { 9 return getState();10 }11 12 final int nonfairTryAcquireShared(int acquires) {13 for (;;) {14 int available = getState();15 int remaining = available - acquires;16 if (remaining < 0 ||17 compareAndSetState(available, remaining))18 return remaining;19 }20 }21 22 protected final boolean tryReleaseShared(int releases) {23 for (;;) {24 int current = getState();25 int next = current + releases;26 if (next < current) // overflow27 throw new Error("Maximum permit count exceeded");28 if (compareAndSetState(current, next))29 return true;30 }31 }32 33 final void reducePermits(int reductions) {34 for (;;) {35 int current = getState();36 int next = current - reductions;37 if (next > current) // underflow38 throw new Error("Permit count underflow");39 if (compareAndSetState(current, next))40 return;41 }42 }43 44 final int drainPermits() {45 for (;;) {46 int current = getState();47 if (current == 0 || compareAndSetState(current, 0))48 return current;49 }50 }51 }和ReentrantLock的Sync差不多，Semaphore的Sync定义了以下的一些主要内容：
getPermits方法获取当前的许可剩余量还剩多少，即还有多少线程可以同时获得信号量
定义了非公平信号量获取共享锁的逻辑nonfairTryAcquireShared
定义了公平模式释放信号量的逻辑tryReleaseShared，相当于释放一次信号量，state就向上+1（信号量每次的获取与释放都是以1为单位的）
再看下公平信号量的实现，同样的FairSync，继承自Sync，代码为：
1 static final class FairSync extends Sync { 2 private static final long serialVersionUID = 2014338818796000944L; 3 4 FairSync(int permits) { 5 super(permits); 6 } 7 8 protected int tryAcquireShared(int acquires) { 9 for (;;) {10 if (hasQueuedPredecessors())11 return -1;12 int available = getState();13 int remaining = available - acquires;14 if (remaining < 0 ||15 compareAndSetState(available, remaining))16 return remaining;17 }18 }19 }首先第10行的hasQueuedPredecessors方法，前面已经说过了，如果已经有了FIFO队列或者当前线程不是FIFO队列中在等待的第一条线程，返回-1，表示无法获取共享锁成功。
接着获取available，available就是state，用volatile修饰，所以线程中可以看到最新的state，信号量的acquires是1，每次获取信号量都对state-1，两种情况直接返回：
remaining减完<0
通过cas设置成功
之后就是和之前说过的共享锁的逻辑了，如果返回的是一个<0的数字，那么构建FIFO队列，线程阻塞，直到前面的执行完才能唤醒后面的。
接着看一下非公平信号量的实现，NonfairSync继承Sync：
1 static final class NonfairSync extends Sync { 2 private static final long serialVersionUID = -2694183684443567898L; 3 4 NonfairSync(int permits) { 5 super(permits); 6 } 7 8 protected int tryAcquireShared(int acquires) { 9 return nonfairTryAcquireShared(acquires);10 }11 }nonfairTryAcquireShared在父类已经实现了，再贴一下代码：
1 final int nonfairTryAcquireShared(int acquires) {2 for (;;) {3 int available = getState();4 int remaining = available - acquires;5 if (remaining < 0 ||6 compareAndSetState(available, remaining))7 return remaining;8 }9 }看到这里和公平Semaphore只有一点差别：不会前置进行一次hasQueuedPredecessors()判断。即当前有没有构建为一个FIFO队列，队列里面第一个等待的线程是不是自身都无所谓，对于非公平Semaphore都一样，反正线程调用Semaphore的acquire方法就将当前state-1，如果得到的remaining设置成功或者CAS操作成功就返回，这种操作没有遵循先到先得的原则，即非公平信号量。
至于非公平信号量对比公平信号量的优点，和ReentrantLock的非公平锁对比ReentrantLock的公平锁一样，就不说了。
CountDownLatch实现原理
CountDownLatch即计数器自减的一种闭锁，某线程阻塞，对一个计数器自减到0，此线程被唤醒，CountDownLatch具体用法可见Java多线程19：多线程下的其他组件之CountDownLatch、Semaphore、Exchanger。
CountDownLatch是一种共享锁，通过await()方法与countDown()两个方法实现自身的功能，首先看一下await()方法的实现：
1 public void await() throws InterruptedException { 2 sync.acquireSharedInterruptibly(1); 3 }acquireSharedInterruptibly最终又回到tryAcquireShared方法上，直接贴整个Sync的代码实现：
1 private static final class Sync extends AbstractQueuedSynchronizer { 2 private static final long serialVersionUID = 4982264981922014374L; 3 4 Sync(int count) { 5 setState(count); 6 } 7 8 int getCount() { 9 return getState();10 }11 12 protected int tryAcquireShared(int acquires) {13 return (getState() == 0) ? 1 : -1;14 }15 16 protected boolean tryReleaseShared(int releases) {17 // Decrement count; signal when transition to zero18 for (;;) {19 int c = getState();20 if (c == 0)21 return false;22 int nextc = c-1;23 if (compareAndSetState(c, nextc))24 return nextc == 0;25 }26 }27 }其实看到tryAcquireShared方法，理解AbstractQueuedSynchronizer共享锁原理的，不用看countDown方法应该都能猜countDown方法是如何实现的。我这里总结一下：
传入一个count，state就等于count，await的时候判断是不是0，是0返回1表示成功，不是0返回-1表示失败，构建FIFO队列，head头只连接一个Node，Node中的线程就是调用CountDownLatch的await()方法的线程
每次countDown的时候对state-1，直到state减到0的时候才算tryReleaseShared成功，tryReleaseShared成功，唤醒被挂起的线程
为了验证（2），看一下上面Sync的tryReleaseShared方法就可以了，确实是这么实现的。
위 내용은 AbstractQueuedSynchronizer 기반 동시성 클래스에 대한 자세한 설명의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!