ReentrantLock原理之公平锁原理

1 公平锁

static final class FairSync extends Sync {
 private static final long serialVersionUID = -3000897897090466540L;
 final void lock() {
     acquire(1);
 }
 
 // AQS 继承的方法
 public final void acquire(int arg) {
     if (
     !tryAcquire(arg) &&
     acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
     ) {
         selfInterrupt();
     }
 }
 // 与非公平锁主要区别在于 tryAcquire 方法的实现
 protected final boolean tryAcquire(int acquires) {
     final Thread current = Thread.currentThread();
     int c = getState();
     if (c == 0) {
         // 先检查 AQS 队列中是否有前驱节点, 没有才去竞争
         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");
             setState(nextc);
             return true;
         }
         return false;
     }
 
 //1 AQS 继承的方法
 public final boolean hasQueuedPredecessors() {
     Node t = tail;
     Node h = head;
     Node s;
     // h != t 时表示队列中有 Node
     return h != t &&
     (
     // (s = h.next) == null 表示队列中还有没有老二
     (s = h.next) == null ||
     // 或者队列中老二线程不是此线程
     s.thread != Thread.currentThread()
     );
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

2 可重入

static final class NonfairSync extends Sync {
     // ...
 
     // Sync 继承的方法
 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()) {
         // state++
         int nextc = c + acquires;
         if (nextc < 0) // overflow
         throw new Error("Maximum lock count exceeded");
         setState(nextc);
         return true;
     }
     return false;
 }
 
     // Sync 继承方法
     protected final boolean tryRelease(int releases) {
         // state-- 
         int c = getState() - releases;
         if (Thread.currentThread() != getExclusiveOwnerThread())
         throw new IllegalMonitorStateException();
         boolean free = false;
         // 支持锁重入, 只有 state 减为 0, 才释放成功
         if (c == 0) {
             free = true;
             setExclusiveOwnerThread(null);
         }
         setState(c);
         return free;
     }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

3 可打断

不可打断原理

即使它被打断，仍会驻留在 AQS 队列中，一直要等到获得锁后方能得知自己被打断了.

// Sync 继承自 AQS
static final class NonfairSync extends Sync {
 // ...
 
 private final boolean parkAndCheckInterrupt() {
     // 如果打断标记已经是 true, 则 park 会失效
     LockSupport.park(this);
     // interrupted 会清除打断标记
     return Thread.interrupted();
 }
 
 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;
                 failed = false;
                 // 还是需要获得锁后, 才能返回打断状态
                 return interrupted;
             }
             if (
             shouldParkAfterFailedAcquire(p, node) &&
             parkAndCheckInterrupt()
             ) {
                 // 如果是因为 interrupt 被唤醒, 返回打断状态为 true
                 interrupted = true;
             }
         }
     } finally {
         if (failed)
         cancelAcquire(node);
     }
 }
 
 public final void acquire(int arg) {
          if (
     !tryAcquire(arg) &&
     acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
     ) {
         // 如果打断状态为 true
         selfInterrupt();
     }
 }
 
     static void selfInterrupt() {
         // 重新产生一次中断
         Thread.currentThread().interrupt();
     }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

可打断模式

static final class NonfairSync extends Sync {
 public final void acquireInterruptibly(int arg) throws InterruptedException {
     if (Thread.interrupted())
     throw new InterruptedException();
     // 如果没有获得到锁, 进入 1
     if (!tryAcquire(arg))
         doAcquireInterruptibly(arg);
     }
 
 // 1 可打断的获取锁流程
 private void doAcquireInterruptibly(int arg) throws InterruptedException {
     final Node node = addWaiter(Node.EXCLUSIVE);
     boolean failed = true;
     try {
         for (;;) {
             final Node p = node.predecessor();
             if (p == head && tryAcquire(arg)) {
             setHead(node);
             p.next = null; // help GC
             failed = false;
             return;
             }
             if (shouldParkAfterFailedAcquire(p, node) &&
                 parkAndCheckInterrupt()) {
                 // 在 park 过程中如果被 interrupt 会进入此
                 // 这时候抛出异常, 而不会再次进入 for (;;)
                 throw new InterruptedException();
             }
         }
     } finally {
         if (failed)
         cancelAcquire(node);
     }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

4条件变量

每个条件变量其实就对应着一个等待队列，其实现类是 ConditionObject

await 流程

开始 Thread-0 持有锁，调用 await，进入 ConditionObject 的 addConditionWaiter 流程 创建新的 Node 状态为 -2（Node.CONDITION），关联 Thread-0，加入等待队列尾部

接下来进入 AQS 的 fullyRelease 流程，释放同步器上的锁

unpark AQS 队列中的下一个节点，竞争锁，假设没有其他竞争线程，那么 Thread-1 竞争成功

park 阻塞 Thread-0

signal 流程

如Thread-1 要来唤醒 Thread-0

进入 ConditionObject 的 doSignal 流程，取得等待队列中第一个 Node，即 Thread-0 所在 Node

执行 transferForSignal 流程，将该 Node 加入 AQS 队列尾部，将 Thread-0 的 waitStatus 改为 0，Thread-3 的 waitStatus 改为 -1

Thread-1 释放锁，进入 unlock 流程.