Java中AQS基本实现原理

一、AQS概述

AQS全名AbstractQueuedSynchronizer,意为抽象队列同步器,JUC(java.util.concurrent包)下面的Lock和其他一些并发工具类都是基于它来实现的。AQS维护了一个volatile的state和一个CLH(FIFO)双向队列。

二、分析

state

state是一个由volatile修饰的int变量,它的访问方式有三种:

  • getState()
  • setState(int newState)
  • compareAndSetState(int expect, int update)
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
/**
* 由volatile修饰的state
*/
private volatile int state;

/**
* 基于内存可见性的读
*/
protected final int getState() {
return state;
}

/**
* 基于内存可见性的写
*/
protected final void setState(int newState) {
state = newState;
}

/**
* 使用CAS+volatile,基于原子性与可见性的对state进行设值
*/
protected final boolean compareAndSetState(int expect, int update) {
// 使用Unsafe类,调用JNI方法
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

资源获取主要有两种形式:

  • 独占式(EXCLUSIVE)

仅有一个线程能在同一时刻获取到资源并处理,如ReentrantLock的实现。

  • 共享式(SHARED)

多个线程可以同时获取到资源并处理,如Semaphore/CountDownLatch等。

AQS中大部分逻辑已经被实现,集成类只需要重写state的获取(acquire)与释放(release)方法,因为在AQS中,这些方法默认定义的实现方式都是抛出不支持操作异常,所以按需实现即可。

其中需要继承类重写的方法有:

  • tryAcquire(int arg)

此方法是独占式的获取资源方法,成功则返回true,失败返回false。

  • tryRelease(int arg)

此方法是独占式的释放资源方法,成功则返回true,失败返回false。

  • tryAcquireShared(int arg)

此方法是共享式的获取资源方法,返回负数表示失败,0表示获取成功,但是没有可用资源,正数表示获取成功,且有可用资源。

  • tryReleaseShared(int arg)

此方法是共享式的释放资源方法,如果允许唤醒后续等待线程则返回true,不允许则返回false。

  • isHeldExclusively()

判断当前线程是否正在独享资源,是则返回true,否则返回false。

CLH(FIFO)队列

AQS中是通过内部类Node来维护一个CLH队列的。源码如下:

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
static final class Node {
/** 标记共享式访问 */
static final Node SHARED = new Node();
/** 标记独占式访问 */
static final Node EXCLUSIVE = null;

/** 字段waitStatus的值,表示当前节点已取消等待 */
static final int CANCELLED = 1;
/**字段waitStatus的值,表示当前节点取消或释放资源后,通知下一个节点 */
static final int SIGNAL = -1;
/** 表示正在等待触发条件 */
static final int CONDITION = -2;
/**
* 表示下一个共享获取应无条件传播
*/
static final int PROPAGATE = -3;

/**
* Status field, taking on only the values:
* SIGNAL: 表示当前节点取消或释放资源后,通知下一个节点
* CANCELLED: 表示当前节点已取消等待
* CONDITION: 表示正在等待触发条件
* PROPAGATE: 表示下一个共享获取应无条件传播
*/
volatile int waitStatus;

/**
* 前节点
*/
volatile Node prev;

/**
* 下一个节点
*/
volatile Node next;

/**
* 节点对应线程
*/
volatile Thread thread;

/**
* 下一个等待的节点
*/
Node nextWaiter;

/**
* 是否是共享式访问
*/
final boolean isShared() {
return nextWaiter == SHARED;
}

/**
* 返回前节点
*/
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}

Node() { // 共享式访问的构造函数
}

Node(Thread thread, Node mode) { // 用于被添加等待者使用
this.nextWaiter = mode;
this.thread = thread;
}

Node(Thread thread, int waitStatus) { // 用于Condition使用
this.waitStatus = waitStatus;
this.thread = thread;
}
}

独占模式-获取资源

使用AQS中的acquire(int arg)方法

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

该方法分为4个部分:

  • tryAcquire()

需要自己实现的方法,如果获取到资源使用权,则返回true,反之fasle。如果获取到资源,返回true,!true为false,根据&&的短路性,则不会执行后续方法,直接跳过程序。如果未获取到资源,返回false,!false为true,则进入后续方法。

  • addWaiter()

如果未获取到资源使用权,则首先会调用此方法。上源码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private Node addWaiter(Node mode) {
// 封装当前线程和独占模式
Node node = new Node(Thread.currentThread(), mode);
// 获取尾部节点
Node pred = tail;
if (pred != null) {
node.prev = pred;
// CAS设置尾部节点
if (compareAndSetTail(pred, node)) {
// 将为节点的下一节点指向当前node
pred.next = node;
return node;
}
}
// 如果尾结点为空或者设置尾结点失败
enq(node);
return node;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
private Node enq(final Node node) {
// 如果CAS设置未成功则死循环
for (;;) {
// 获得尾结点
Node t = tail;
// 如果尾节点为空,说明CLH队列为空,需要初始化
if (t == null) {
if (compareAndSetHead(new Node()))
tail = head;
} else {
// 设置当前节点的前驱节点
node.prev = t;
// CAS设置当前节点为尾结点
if (compareAndSetTail(t, node)) {
// 设置后驱节点
t.next = node;
return t;
}
}
}
}
  • acquiredQueued()
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
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);
// 取消前驱节点(以前的头部)的后节点,方便GC回收
p.next = null; // help GC
// 标识未失败
failed = false;
// 返回中断标志
return interrupted;
}
// 如果当前节点的前驱节点不是头结点或获取资源失败
// 需要用shouldParkAfterFailedAcquire函数判断是否需要阻塞该节点持有的线程
// 如果需要阻塞,则执行parkAndCheckInterrupt方法,并设置被中断
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
// 如果最终获取资源失败
if (failed)
// 当前节点取消获取资源
cancelAcquire(node);
}
}
  • selfInterrupt()

中断当前线程

1
2
3
static void selfInterrupt() {
Thread.currentThread().interrupt();
}

独占模式-释放资源

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;
// 头结点不为空且等待状态值不为0
if (h != null && h.waitStatus != 0)
// 唤醒后续等待线程
unparkSuccessor(h);
return true;
}
return false;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
// 如果等待状态值小于0
if (ws < 0)
// 使用CAS将waitStatus设置为0
compareAndSetWaitStatus(node, ws, 0);

Node s = node.next;
// 如果当前节点没有后继节点或者后继节点放弃竞争资源
if (s == null || s.waitStatus > 0) {
s = null;
// 从队列尾部循环直到当前节点,找到最近的且等待状态值小于0的节点
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
// 如果找到的后继节点不为空,则唤醒其持有的线程
if (s != null)
LockSupport.unpark(s.thread);
}
查看评论