ReentrantLock原理之非公平锁

下面是ReentrantLock关系类图:

1 非公平锁

1 加锁流程

public ReentrantLock() {
    // 默认非公平锁
     sync = new NonfairSync();
}
1
2
3
4

NonfairSync 继承自 AQS.

没有竞争时

第一个竞争时

说明:

1 CAS 尝试将 state 由 0 改为 1，结果失败

2 进入 tryAcquire 逻辑，这时 state 已经是1，结果仍然失败

3 进入 addWaiter 逻辑，构造 Node 队列

• 黄色三角表示该 Node 的 waitStatus 状态，其中 0 为默认正常状态
• Node 的创建是懒惰的
• 第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程
  

当前线程进入acquireQueued 逻辑

• acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 park 阻塞
• 如果是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
• 进入 shouldParkAfterFailedAcquire 逻辑，将前驱 node，即 head 的 waitStatus 改为 -1，这次返回 false

• shouldParkAfterFailedAcquire 执行完毕回到 acquireQueued ，再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
• 再次进入 shouldParkAfterFailedAcquire 时，这时因为其前驱 node 的 waitStatus 已经是 -1，这次返回 true
• 进入 parkAndCheckInterrupt， Thread-1 park

多个线程竞争时

多个线程竞争失败，变成如下图列

2 解锁流程

Thread-0 释放锁，进入 tryRelease 流程，如果成功,会做两件事

• 设置 exclusiveOwnerThread 为 null
• state = 0

当前队列不为 null，并且 head 的 waitStatus = -1，进入 unparkSuccessor 流程

找到队列中离 head 最近的一个 Node（没取消的），unpark 恢复其运行，即为 Thread-1

Thread-1 的 acquireQueued 流程:

如果加锁成功（没有竞争），会设置:

• 如果加锁成功（没有竞争），会设置
• head 指向刚刚 Thread-1 所在的 Node，该 Node 清空 Thread
• 原本的 head 因为从链表断开，而可被垃圾回收

非公平竞争

此时有其他线程Thread-4来竞争,且被竞争成功.

• Thread-4 被设置为 exclusiveOwnerThread，state = 1
• Thread-1 再次进入 acquireQueued 流程，获取锁失败，重新进入 park 阻塞

加锁源码

// Sync 继承自 AQS
static final class NonfairSync extends Sync {
 private static final long serialVersionUID = 7316153563782823691L;
 
 // 加锁实现
 final void lock() {
 // 首先用 cas 尝试（仅尝试一次）将 state 从 0 改为 1, 如果成功表示获得了独占锁
 if (compareAndSetState(0, 1))
     setExclusiveOwnerThread(Thread.currentThread());
 else
     // 如果尝试失败，进入 ㈠
     acquire(1);
 }
 
 // 1 AQS 继承过来的方法, 方便阅读, 放在此处
 public final void acquire(int arg) {
 // 2 tryAcquire 
 if (
     !tryAcquire(arg) &&
     // 当 tryAcquire 返回为 false 时, 先调用 addWaiter , 接着 acquireQueued 
     acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
 ) {
     selfInterrupt();
 }
 }
 
 // 2
 protected final boolean tryAcquire(int acquires) {
 return nonfairTryAcquire(acquires);
 }
 
 // 3
 final boolean nonfairTryAcquire(int acquires) {
 final Thread current = Thread.currentThread();
 int c = getState();
 // 如果还没有获得锁
 if (c == 0) {
     // 尝试用 cas 获得, 这里体现了非公平性: 不去检查 AQS 队列
     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;
 }
 
 // 4 AQS 继承过来的方法, 方便阅读, 放在此处
 private Node addWaiter(Node mode) {
    
     // 将当前线程关联到一个 Node 对象上, 模式为独占模式
     Node node = new Node(Thread.currentThread(), mode);
     // 如果 tail 不为 null, cas 尝试将 Node 对象加入 AQS 队列尾部
     Node pred = tail;
     if (pred != null) {
         node.prev = pred;
         if (compareAndSetTail(pred, node)) {
             // 双向链表
             pred.next = node;
             return node;
         }
     }
     // 尝试将 Node 加入 AQS, 进入 ㈥
     enq(node);
     return node;
 }
 
    // 6
 private Node enq(final Node node) {
     for (;;) {
         Node t = tail;
         if (t == null) {
             // 还没有, 设置 head 为哨兵节点（不对应线程，状态为 0）
             if (compareAndSetHead(new Node())) {
                 tail = head;
             }
         } else {
             // cas 尝试将 Node 对象加入 AQS 队列尾部
             node.prev = t;
             if (compareAndSetTail(t, node)) {
                 t.next = node;
                 return t;
             }
       	}
    }
 }
 
 // 5
 final boolean acquireQueued(final Node node, int arg) {
     boolean failed = true;
     try {
         boolean interrupted = false;
         for (;;) {
             final Node p = node.predecessor();
             // 上一个节点是 head, 表示轮到自己（当前线程对应的 node）了, 尝试获取
             if (p == head && tryAcquire(arg)) {
                 // 获取成功, 设置自己（当前线程对应的 node）为 head
                 setHead(node);
                 // 上一个节点 help GC
                 p.next = null;
                 failed = false;
                 // 返回中断标记 false
                 return interrupted; 
             }
             if (
             // 判断是否应当 park, 进入 ㈦
             shouldParkAfterFailedAcquire(p, node) &&
             // park 等待, 此时 Node 的状态被置为 Node.SIGNAL ㈧
             parkAndCheckInterrupt()
             ) {
                 interrupted = true;
             }
         }
     } finally {
         if (failed)
         cancelAcquire(node);
     }
 }
 
    // 7
 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
     // 获取上一个节点的状态
     int ws = pred.waitStatus;
     if (ws == Node.SIGNAL) {
         // 上一个节点都在阻塞, 那么自己也阻塞好了
         return true;
     }
     
     // > 0 表示取消状态
     if (ws > 0) {
         // 上一个节点取消, 那么重构删除前面所有取消的节点, 返回到外层循环重试
         do {
             node.prev = pred = pred.prev;
         } while (pred.waitStatus > 0);
             pred.next = node;
         } else {
             // 这次还没有阻塞
             // 但下次如果重试不成功, 则需要阻塞，这时需要设置上一个节点状态为 Node.SIGNAL
             compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
         }
         return false;
     }
 
     // 8 阻塞当前线程
     private final boolean parkAndCheckInterrupt() {
         LockSupport.park(this);
         return Thread.interrupted();
     }
}
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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

是否需要 unpark 是由当前节点的前驱节点的 waitStatus == Node.SIGNAL 来决定，而不是本节点的 waitStatus 决定

解锁源码

// Sync 继承自 AQS
static final class NonfairSync extends Sync {
 // 解锁实现
 public void unlock() {
 sync.release(1);
 }
 
 public final boolean release(int arg) {
     // 尝试释放锁, 进入 ㈠
     if (tryRelease(arg)) {
         // 队列头节点 unpark
         Node h = head; 
         if (
             // 队列不为 null
             h != null &&
             // waitStatus == Node.SIGNAL 才需要 unpark
             h.waitStatus != 0
         ) {
             // unpark AQS 中等待的线程, 进入 ㈡
             unparkSuccessor(h);
         }
         return true;
     }
     return false;
 }
 
 // 1 
 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;
 }
 
 // 2 AQS 
 private void unparkSuccessor(Node node) {
     // 如果状态为 Node.SIGNAL 尝试重置状态为 0
     // 不成功也可以
     int ws = node.waitStatus;
     if (ws < 0) {
         compareAndSetWaitStatus(node, ws, 0);
     }
 // 找到需要 unpark 的节点, 但本节点从 AQS 队列中脱离, 是由唤醒节点完成的
      Node s = node.next;
 // 不考虑已取消的节点, 从 AQS 队列从后至前找到队列最前面需要 unpark 的节点
 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);
     }
}
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
54
55
56
57
58
59
60
61
62
63