Java多线程学习（四）—— 队列同步器AQS

发表于 | 分类于

AQS 简介

AbstractQueuedSynchronizer，简称 AQS，队列同步器，是 java.util.concurrent 包里的一个重要组件。Java 并发编程中常用的锁及其他同步组件的基础均为 AQS 。其内部使用一个 int 变量表示同步状态。AQS 通常被继承使用，即使用者需要自己定义变更同步状态的相关操作方法。这些方法会被 AQS 中的模板方法调用，然后锁及其他同步组件，通过调用AQS中已经实现的模板方法（如：acquire(int) 、 acquireInterruptibly(int) 、 release(int) 等），来实现锁或者其他同步组件的功能。

从一个简单自定义锁开始

一个简易互斥锁的实现

文章的开篇，我们通过编写一个实现 Lock 接口且内部持有一个自定义同步器的互斥锁的实例，来简单的了解 AQS 的简单工作流程。具体示例代码如下：

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
/**
 * @description: 简单自定义Lock的实现
 * @author: jayden
 * @date: 2017/12/11
 */
public class MutexLock implements Lock {
	//自定义的锁持有一个同步器，用于同步管理
    private final Synchronizer synchronizer = new Synchronizer();
    @Override
    public void lock() {
        synchronizer.acquire(1);
    }
    @Override
    public void lockInterruptibly() throws InterruptedException {
        synchronizer.acquireInterruptibly(1);
    }
    @Override
    public boolean tryLock() {
        return synchronizer.tryAcquire(1);
    }
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return synchronizer.tryAcquireNanos(1, unit.toNanos(time));
    }
    @Override
    public void unlock() {
        synchronizer.release(1);
    }
    @Override
    public Condition newCondition() {
        return synchronizer.newCondition();
    }
    private static class Synchronizer extends AbstractQueuedSynchronizer {
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }
        @Override
        public boolean tryAcquire(int acquires) {
            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;
        }
        Condition newCondition() {
            return new ConditionObject();
        }
    }
}

以上代码完成了一个简易互斥锁的实现。像 AQS 作者的推荐用法那样，我们自定义锁的内部持有了一个自定义的 AQS 子类，该子类主要覆写了 AQS 定义中未进行实现的三个方法：isHeldExclusively() 、tryAcquire(int) 、tryRelease(int)。然后在 Lock 接口定义的方法中调用 AQS 的方法，便可以实现锁的功能。

再看示例实现

从上一个示例代码中，我们可以看出，该简易互斥锁的基本实现中，主要分为两大部分：

  • 实现 Lock 接口中定义的方法，供锁的使用者调用；主要有 lock()、unlock() 等。Lock 接口方法中通过调用同步器已经实现的模板方法，来实现锁的功能。
  • 内部实现并持有一个 AQS 的子类，该子类中需要重写 AQS 默认实现为抛出 UnsupportedOperationException 异常的方法。在本例中主要实现了独立式获取和释放同步状态的方法；有 tryAcquire(int)、tryRelease(int)以及 isHeldExclusively() 方法。

AQS 基本介绍及使用

同步器 AQS 的基本使用

同步器的使用方式通常是组合在自定义同步组件的实现中。同步器中有一些被指定 需要重写的方法，还有一些 模板方法。在自定义同步组件时，该同步组件内部持有自定义同步组件的实例。同步组件 的方法通过调用其内部持有的 同步器 的模板方法实现功能；而这些模板方法，则是我们在同步器中重写的方法的调用者。
同步器中定义的需要重写的方法如下：

  • protected boolean tryAcquire(int arg)：独占式获取同步状态
  • protected boolean tryRelease(int arg)：独占式释放同步状态
  • protected int tryAcquireShared(int arg)：共享式获取同步状态
  • protected boolean tryReleaseShared(int arg)：共享式释放同步状态
  • protected boolean isHeldExclusively()：该同步器是否被当前线程所独占
    重写同步器方法的时候访问和修改同步状态用到的方法：
  • getState()：获取当前同步状态
  • setState(int state)：设置当前同步状态
  • compareAndSet(int exp, int update)：通过使用 CAS 来设置当前状态
    同步器中定义的模板方法主要有三类：
  • 独占式获取、释放同步状态
  • 共享式获取、释放同步状态
  • 查询同步队列中的等待线程情况
    有了这些对 AQS 的基础了解后，我们便可以自己实现一个小的同步组件，同时了解 AQS 的实现方式对于进一步对 java.util.concurrent 包内组件的解读也有着很大的好处。

再看上文示例

了解了AQS的基本使用方式后，我们再来看上文给出的示例。此处仅针对自定义同步器部分进行介绍。
根据上部分的介绍，我们知道实现自定义同步组件的第一步便是需要重写同步器中的几个方法。由于我们需要实现的是一个独占式同步组件，所以需要覆写 tryAcquire(int arg) 、 tryRelease(int arg) 、 isHeldExclusively() 这三个方法。

自定义 tryAcquire(int acquires)

自定义 tryAcquire(int acquires) 方法的代码块如下：

1
2
3
4
5
6
7
8
@Override
public boolean tryAcquire(int acquires) {
	if (compareAndSetState(0, 1)) {
		setExclusiveOwnerThread(Thread.currentThread());
        return true;
    }
    return false;
}

由上文代码中可以看出，我们的这个同步组件在获取同步状态的时候，仅需要通过CAS不断尝试去将state位置为1，然后将当前线程设置为独占线程，即完成了获取同步状态的功能。该方法会被 void acquire(int arg) 方法所调用，而 void acquire(int arg) 这一模板方法将在获取锁的时候调用，用以完成加锁功能。

自定义 tryRelease(int arg)

自定义 tryRelease(int release) 方法的代码块如下：

1
2
3
4
5
6
7
8
9
@Override
protected boolean tryRelease(int release) {
    if (getState() == 0) {
        throw new IllegalMonitorStateException();
    }
    setExclusiveOwnerThread(null);
    setState(0);
    return true;
}

该方法用于释放掉同步状态，由代码中可以看出，我们只需要释放掉当前线程对该同步器的占用，同时将state复位为0，即可释放掉该同步状态，达到释放锁的目的。

自定义 isHeldExclusively()

自定义 isHeldExclusively() 方法的代码块如下：

1
2
3
4
@Override
protected boolean isHeldExclusively() {
    return getState() == 1;
}

该方法用于获取当前同步状态。此处仅实现为根据 state 位状态返回当前线程是否独占同步器的结果，供判断条件调用。

通过再次对代码的理解，应该已经大概了解了一个简单锁的实现。本篇中并没有针对 Lock 接口进行过多的介绍，仅是对 AQS 的扩展使用进行了一个简单的介绍。

总结

本文仅是针对 AbstractQueuedSynchronizer 同步器的一个简单入门了解。随后还会深入理解 AQS 的具体实现细节，但有了此篇的介绍，在阅读 juc 包里的源码时，也会有一定的帮助。
由于初学且行文时间较赶，如果错误还烦请大家多多指教。有相关讨论可以联系 Email：JaydenRansom@outlook.com