AQS同步器

AQS同步器

  在 java.util.concurrent (JUC) 并发包中,如 ReentrantLock,Semaphore,CountDownLatch 等并发类的同步控制都是基于 AbstractQueuedSynchronizer (简称AQS) 这个同步器抽象类来实现的。在这里较为深入的讨论同步器抽象类的实现原理与应用。

AQS简介

AbstractQueuedSynchronizer内部维护着一个FIFO的CLH队列,队列中的每个Node代表着一个需要获取锁的线程

  自旋锁:自旋锁是指当一个线程尝试获取某个锁时,如果该锁已被其他线程占用,就一直循环检测锁是否被释放,而不是立刻进入线程挂起或睡眠状态。

  • CLH锁(Craig, Landin, and Hagersten locks):基于链表的可扩展、高性能、公平的自旋锁,它不断轮询前驱的状态,如果发现前驱释放了锁就结束自旋
  • MCS锁:在当前结点自旋,但由前驱结点通知其结束自旋

ASQ采用的是一种变种的CLH队列锁:原始CLH是在前驱结点自旋,通过判断pred.locked来自旋,而AQS的CLH则是根据前驱结点的状态来控制阻塞,不会一直自旋。同时当前驱结点释放锁时会去唤醒该结点使其参与竞争锁。 AQS的结点的定义如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
static final class Node {

volatile int waitStatus;

volatile Node prev;

volatile Node next;

volatile Thread thread;

// 指向condition队列中的后继节点
Node nextWaiter;
}

Node结点中分别有指向前驱,后继的结点,入队时的线程以及结点状态(Condition队列本文不涉及)。结点状态会存在以下几种:

  • CANCELLED:线程取消
  • SIGNAL:当前线程的后继线程被阻塞或者即将被阻塞,当前线程释放锁或者取消后需要唤醒后继线程
  • CONDITION:在等待 Condition ,也就是在condition队列中
  • PROPAGATE:当头结点处于PROPAGATE,需要唤醒后继线程,为了保证共享模式下唤醒机制正常
  • 0:初始状态

基于上述Node的定义,AQS基本属性如下:

1
2
3
4
5
6
7
8
// 队列的头结点
private transient volatile Node head;

// 队列的尾节点
private transient volatile Node tail;

// 同步状态
private volatile int state;

API

AbstractQueuedSynchronizer的提供的接口主要有两种类型

控制同步状态

AbstractQueuedSynchronizer并不实现同步接口,所有对同步状态的控制都交由子类同步组件控制。比如tryAcquire代表由子类控制当前线程是否能独占式获取同步状态成功

方法 说明
boolean tryAcquire(int arg) 独占式获取同步状态
boolean tryRelease(int arg) 独占式释放同步状态
int tryAcquireShared(int arg) 共享式获取同步状态
boolean tryReleaseShared(int arg) 共享式私房同步状态
boolean isHeldExclusively() 检测当前线程是否获取独占锁

而在多线程环境中对状态的操纵必须确保原子性,因此它还提供了对状态控制的三组API:

方法 说明
int getState() 获取同步状态
void setState() 设置同步状态
boolean compareAndSetState(int expect, int update) 通过 CAS 设置同步状态

通过这三组API,子类可以线程安全的控制同步状态(同时子类需要确保实现是非阻塞的)

模板方法

模板方法封装了获取同步状态成功或失败后的在队列中的一系列操作,子类可以直接调用

方法 说明
void acquire(int arg) 独占式获取同步状态,该方法将会调用 tryAcquire 尝试获取同步状态。获取成功则返回,获取失败,线程进入同步队列等待。
void acquireInterruptibly(int arg) 响应中断版的 acquire
boolean tryAcquireNanos(int arg,long nanos) 超时+响应中断版的 acquire
void acquireShared(int arg) 共享式获取同步状态,同一时刻可能会有多个线程获得同步状态。比如读写锁的读锁就是就是调用这个方法获取同步状态的。
void acquireSharedInterruptibly(int arg) 响应中断版的 acquireShared
boolean tryAcquireSharedNanos(int arg,long nanos) 超时+响应中断版的 acquireShared
boolean release(int arg) 独占式释放同步状态
boolean releaseShared(int arg) 共享式释放同步状态

互斥锁

acquire

1
2
3
4
5
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}

tryAcquire方法代表尝试获取一次互斥锁,需要子类根据需求去实现(比如ReentrantLock实现了公平锁和非公平锁),通过布尔变量来标志获取状态:

1
2
3
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}

若获取失败,则通过addWaiter方法将当前线程添加至阻塞队列

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
private Node addWaiter(Node mode) {

// 将线程封装在Node节点中
Node node = new Node(Thread.currentThread(), mode);

// CAS尝试将该节点插在队列尾
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}

// 如果不成功则通过自旋的方式插到队尾,直到插入成功
enq(node);
return node;
}

private Node enq(final Node node) {
for (;;) {
Node t = tail;
// 设置头结点,初始情况下,头结点是一个空结点(这里不会直接返回,因此即使阻塞队列为空,当前节点仍然是插在空结点之后)
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
// 插入该结点到队尾
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}

插入完成后,则会调用acquireQueued()方法对该结点进行有限次自旋获取锁,并在到达边界条件后阻塞

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
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 GC
failed = false;
return interrupted;
}

// 判断线程需不需要阻塞,和CLH不同,线程并不总是参与竞争锁,而是仅当线程被唤醒时竞争锁
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
// 如果前驱节点为SIGNAL状态,那么在释放锁时会唤醒后继结点
// 因此这种情况当前结点会阻塞自己
if (ws == Node.SIGNAL)
return true;

// 如果前驱节点为CANCELLED状态,那么从后向前找到第一个非取消状态的节点
// 并更新当前结点的前驱为该结点
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {

// 如果前驱节点为0或PROPAGATE,那么设置前驱结点的状态为SIGNAL(可以说这一步才是标志会将每一个节点阻塞的一步)
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}

// LockSupport.park(this)来挂起线程,然后就停在这里了,等待被唤醒
// 返回的时候会先判断是否由线程中断造成的,如果由线程中断造成,在这里会接下去置中断标记
// 而lockInterruptibly方法则是抛出异常
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}

那么总结下acquire方法的逻辑:

  1. 尝试获取互斥锁,若获取成功则直接返回
  2. 若获取失败,则将当前线程添加到阻塞队列尾(CAS操作插入,自旋直到插入成功为止)
  3. 自旋/阻塞获取锁
    1. 尝试获取互斥锁(前驱结点必须为头结点时,当前结点才有资格竞争锁),若获取成功则将当前结点设为头结点后退出
    2. 若前驱结点为SIGNAL状态,则阻塞当前结点(唤醒后继续循环 自旋/阻塞获取锁
    3. 若前驱结点为CANCELLED状态,则更新前驱到非取消结点
    4. 若前驱结点为0或PROPAGATE,则设置前驱结点状态为SIGNAL状态

release

1
2
3
4
5
6
7
8
9
10
11
12
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;

// h==null的情况就是阻塞队列为空(前面说过,第一个线程持有锁时不会放到头结点中)
// h.waitStatus=0,那么其后的结点必定没有阻塞(前面也说过,因为该值是由后继结点来赋值的,然后仅当该结点状态为阻塞状态,后继结点才会将自己阻塞,即CLH特性,根据前驱结点状态来控制自己)
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}

tryRelease方法代表尝试释放一次互斥锁,需要子类根据需求去实现,通过布尔变量来标志获取状态:

1
2
3
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}

在释放锁成功后,会判断当前结点状态来唤醒后继结点,即当前结点状态为SIGNAL状态时会唤醒后继结点

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);

private void unparkSuccessor(Node node) {

int ws = node.waitStatus;
if (ws < 0)
// 设置头结点状态为0
compareAndSetWaitStatus(node, ws, 0);

// 从队尾往前找,找到waitStatus<=0的所有节点中排在最前面的(>0代表节点取消阻塞)
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);
}

那么总结下release方法的逻辑:

  1. 尝试释放一次互斥锁,若释放失败,则直接返回失败
  2. 释放成功后,唤醒一个后继结点

互斥锁案例

通过以上的理解,可以实现一个简单的互斥锁

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
public class MutexLock implements Lock {

private Sync sync;

public MutexLock() {

this.sync = new Sync();
}

private static class Sync extends AbstractQueuedSynchronizer {

public Sync() {

setState(0);
}

@Override
protected boolean tryAcquire(int acquire) {

if(compareAndSetState(0, 1)){

setExclusiveOwnerThread(Thread.currentThread());
return true;
}

return false;
}

@Override
protected boolean tryRelease(int release) {

if(getState() == 0){
throw new IllegalMonitorStateException();
}

setExclusiveOwnerThread(null);
setState(0);
return true;

}
}

@Override
public void lock() {

sync.acquire(1);
}

@Override
public void lockInterruptibly() throws InterruptedException {

sync.acquireInterruptibly(1);
}

@Override
public boolean tryLock() {

return sync.tryAcquire(1);
}

@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {

return sync.tryAcquireNanos(1, unit.toNanos(time));
}

@Override
public void unlock() {

sync.release(1);
}

@Override
public Condition newCondition() {
return null;
}
}

内部类Sync继承AbstractQueuedSynchronizer,并重载tryAcquire和tryRelease方法

  • tryAcquire通过CAS尝试获取一次同步状态(0->1),若获取成功则设置当前持有锁的线程为自己
  • tryRelease判断同步状态是否为1,若是则重置同步状态为0,且设置当前获取锁的线程为null,否则抛出异常(互斥锁的释放不会有并发)

我们可以写个简单的并发计数测试:

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
public class Main {

// 计数
private static int count;

public static void main(String[] args) {

final Mutex mutex = new Mutex();

ExecutorService executorService = new ThreadPoolExecutor(
3, 5, 60, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(100),
new ThreadPoolExecutor.DiscardOldestPolicy()
);

count = 0;

int threadCnt = 10;
for (int i = 0; i < threadCnt; i++){

executorService.execute(new Runnable() {
@Override
public void run() {

for(int i = 0; i < 10000; i++){

mutex.lock();

try {
count++;
}finally {
mutex.unlock();
}
}

}
});
}

executorService.shutdown();

try {
executorService.awaitTermination(1, TimeUnit.DAYS);
} catch (InterruptedException e) {
e.printStackTrace();
}

System.out.println("assert " + threadCnt * 10000 + " = " + count + " is true");
}
}

正常输出为:

1
assert 100000 = 100000 is true

共享锁

acquireShared

1
2
3
4
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}

tryAcquireShared方法尝试获取一次共享锁,需要子类根据需求去实现。但和互斥锁不同的是,它以整型作为状态标志,负数代表获取失败,非负数代表获取成功,0代表成功但之后的竞争线程不会成功

1
2
3
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}

在获取共享锁失败时,会调用doAcquireShared将当前线程添加至阻塞队列并自旋获取共享锁

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
private void doAcquireShared(int arg) {

// 将当前线程添加至阻塞队列
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
// 如果该节点的前一节点为头节点,那么它将有资格参与竞争锁
// 如果获取锁成功,则将当前结点设为头结点
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
// 和互斥锁不同的点,共享锁会在获取锁成功后唤醒后继结点
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
// 判断线程需不需要阻塞
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

它的大体逻辑和互斥锁的自旋获取锁逻辑相同,但是它们之间有个很重要的不同点,即共享锁在获取锁成功后调用 setHeadAndPropagate 来唤醒后继结点

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
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
// 设为头结点
setHead(node);

// 唤醒后继结点
if (propagate > 0 || h == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}

private void doReleaseShared() {

for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
// 如果头结点处于SIGNAL状态,唤醒后继结点
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
// 如果头结点处于0的状态,设置头结点状态为PROPAGATE
// 这是为了解决共享锁的并发唤醒后继结点导致极端情况下存在线程永远无法唤醒的情况
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}

在判断是否需要唤醒后继结点这步,它的判断逻辑是 propagate > 0 || h.waitStatus < 0:

  • propagate > 0 :tryAcquireShared方法的返回值,代表当前线程获取共享锁成功(按理说 propagate = 0 的情况也属于获取锁成功,为什么不加进去呢?这是因为当 propagate = 0 时代表当前已经没有共享资源了,所以唤醒也没有意义了)
  • h.waitStatus < 0 :头结点状态为SIGNAL或PROPAGATE时

  在共享锁中会存在PROPAGATE状态:

  • 获取共享锁成功后,如果头结点状态为0(unparkSuccessor时会将头结点状态设为0),会将头结点状态设为PROPAGATE

    1
    2
    3
    else if (ws == 0 &&
    ​ !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
    continue; // loop on failed CAS
  • 判断后继结点是否需要唤醒时会判断头结点的状态 propagate > 0 || h.waitStatus < 0

之所以需要这个状态是因为共享锁的 唤醒后继结点 操作是并发操作,同时 propagate=0 的情况不会唤醒后继结点,因此在一些极端情况下会存在阻塞结点无法被唤醒的情况

那么我们总结下获取共享锁的逻辑:

  1. 尝试获取共享锁,若获取成功则直接返回
  2. 若获取失败,则将当前线程添加到阻塞队列尾(CAS操作插入,自旋直到插入成功为止)
  3. 自旋/阻塞获取锁
    1. 尝试获取共享锁(前驱结点必须为头结点时,当前结点才有资格竞争锁),若获取成功则尝试唤醒一个后继结点(唤醒的结点如果获取锁成功又会继续唤醒接下去的结点)
    2. 前驱结点为SIGNAL状态,则阻塞当前结点(唤醒后继续循环 自旋/阻塞获取锁
    3. 前驱结点为CANCELLED状态,则更新前驱到非取消结点
    4. 前驱结点为0或PROPAGATE,则设置前驱结点状态为SIGNAL状态

releaseShared

1
2
3
4
5
6
7
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}

tryReleaseShared方法代表尝试释放一次共享锁,需要子类根据需求去实现,通过布尔变量来标志获取状态:

1
2
3
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}

释放成功后会调用doReleaseShared尝试唤醒一个后继结点,上面已经解释了。

共享锁案例

基于以上分析,我们也可以实现一个同时允许N个线程进入的共享锁

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
public class ShareLock implements Lock {

private Sync sync;

public ShareLock(Integer permit) {

this.sync = new Sync(permit);
}

private static class Sync extends AbstractQueuedSynchronizer{

Sync(int permit){

setState(permit);
}

@Override
protected int tryAcquireShared(int acquire) {

for(;;){

int expect = getState();
int update = expect - acquire;

if(update < 0 || compareAndSetState(expect, update)){

return update;
}
}
}

@Override
protected boolean tryReleaseShared(int release) {

for(;;){

int expect = getState();
int update = expect + release;

if(compareAndSetState(expect, update)){

return true;
}
}
}
}

@Override
public void lock() {

sync.acquireShared(1);
}

@Override
public void lockInterruptibly() throws InterruptedException {

sync.acquireSharedInterruptibly(1);
}

@Override
public boolean tryLock() {

return sync.tryAcquireShared(1) >= 0;
}

@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {

return sync.tryAcquireSharedNanos(1, unit.toNanos(time));
}

@Override
public void unlock() {

sync.releaseShared(1);
}

@Override
public Condition newCondition() {
return null;
}
}

接下来对共享锁进行简单的测试:

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
public class Main{

public static void main(String[] args) {

final ShareLock shareLock = new ShareLock(2);

ExecutorService executorService = new ThreadPoolExecutor(
5, 5, 60, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(100),
new ThreadPoolExecutor.DiscardOldestPolicy()
);

int threadCnt = 10;
for (int i = 0; i < threadCnt; i++){

executorService.execute(new Runnable() {
@Override
public void run() {

shareLock.lock();

try {

System.out.println(Thread.currentThread().getName() + ": is running");

Thread.sleep(2000);

} catch (InterruptedException e) {
e.printStackTrace();
} finally {

shareLock.unlock();

}
}
});
}

executorService.shutdown();

try {
executorService.awaitTermination(1, TimeUnit.DAYS);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

测试程序创建了一个允许最多两个线程同时进入的共享锁。因此正常情况下,日志会成双打印。

中断

  • thread.interrupt():中断线程,将会设置该线程的中断状态位,即设置为true(不会中断一个正在运行的线程,而是中断阻塞的线程
  • thread.interrupted():判断某个线程是否已被发送过中断请求,该方法调用后会将中断标示位清除,即重新设置为false
  • Thread.currentThread().isInterrupted():判断某个线程是否已被发送过中断请求,不会将中断标示位清除

如果一个线程处于了阻塞状态(如线程调用了thread.sleep、thread.join、thread.wait、1.5中的condition.await、以及可中断的通道上的 I/O 操作方法后可进入阻塞状态),线程在检查中断标示时如果发现中断标示为true,则会在这些阻塞方法调用处抛出InterruptedException异常,并且在抛出异常后立即将线程的中断标示位清除,即重新设置为false。而如果线程处于非阻塞状态,则需要通过判断Thread.interrupted()或者Thread.isInterrupted()来循环检测

  1. Synchronized在获锁的过程中是不能被中断的,意思是说如果产生了死锁,则不可能被中断

  2. LockSupport的park方法阻塞,能够响应中断,但是不会抛出InterruptedException异常

  3. 一个支持中断线程的程序的标准处理模式

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    public void run() {
    try {

    // do something

    // 1. !Thread.currentThread().isInterrupted()确保在非阻塞时能响应中断
    // 2. try-catch后对InterruptedException处理确保阻塞时对中断进行处理
    while (!Thread.currentThread().isInterrupted()&& more work to do) {
    do more work
    ​ }
    ​ } catch (InterruptedException e) {
    //线程在wait或sleep期间被中断了
    ​ } finally {
    //线程结束前做一些清理工作
    ​ }
    }

在之前所说的 acquire,ascquireShared 方法均不支持中断操作

1
2
3
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;

它们在 LockSupport.park 响应中断后只是置一个中断标记,但是并不会处理,仍然自旋获取锁直到获取成功或阻塞。而 acquireInterruptibly,acquireSharedInterruptibly 方法支持中断操作

1
2
3
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();

它们会在 LockSupport.park 响应中断后抛出 InterruptedException 异常结束线程

参考

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×