【并发编程】并发包中工具类的基础：AQS

2022-08-09 13:53:59

AQS为什么要学？

AQS是jdk并发包java.util.concurrent下绝大部分工具类实现的基础，非常重要！
AQS是工作中并发编程常用的类Lock, Latch, Barrier等，都是基于AQS来实现的！
了解AQS，后面学习一些并发工具类，事半功倍！
AQS是管程模型在java层面的实现！（jvm层面的实现是synchronized）

java层面如何实现管程？

synchronized是自动的加锁解锁，java层面需要手动的加锁解锁。
并发可能有多种场景，但是有一些是相同的，可以把这些抽象出来:AQS。

AQS是什么？

AbstractQueuedSynchronizer类，简称AQS。
抽象同步框架，可以用来实现一个依赖状态的同步器。
java.util.concurrent包中的大多数同步器实现都是围绕着共同的基础行为，比如等待队列、条件队列、独占获取、共享获取....
适用于jdk1.5以上！

AQS的同步等待队列：

主要用于维护获取锁失败时入队的线程。
使用双向链表实现！
线程的使用权可以去实现一个volatile int state，只有一个线程cas 0到1成功后进行执行代码,cas 1到0成功后其他线程开始争抢使用权。
他是一个队列，所以需要有入队、出队操作。
这个锁有可能是独占锁、也可能是共享锁，所以可以将加锁、解锁的实现交给子类。

AQS的条件等待队列：

调用await()的时候会释放锁，然后线程会加入到条件队列，调用signal()唤醒的时候会把条件队列中的线程节点移动到同步队列中，等待再次获得锁。
使用单项链表实现！
需要提供等待唤醒机制：对比synchronized的wait/notify,notifyAll,我们也需要提供等待唤醒的方法await/signal,signalAll。
他是一个队列，所以需要有入队、出队操作。

AQS必须具备的功能

阻塞等待队列
共享/独占
公平/非公平
可重入
允许中断

AQS的内部属性：volatile int state

    /**
     * The synchronization state.
     */
    private volatile int state;

    /**
     * Returns the current value of synchronization state.
     * This operation has memory semantics of a {@code volatile} read.
     * @return current state value
     */
    protected final int getState() {
        return state;
    }

    /**
     * Sets the value of synchronization state.
     * This operation has memory semantics of a {@code volatile} write.
     * @param newState the new state value
     */
    protected final void setState(int newState) {
        state = newState;
    }

    /**
     * Atomically sets synchronization state to the given updated
     * value if the current state value equals the expected value.
     * This operation has memory semantics of a {@code volatile} read
     * and write.
     *
     * @param expect the expected value
     * @param update the new value
     * @return {@code true} if successful. False return indicates that the actual
     *         value was not equal to the expected value.
     */
    // 核心方法，通过cas去改变State的值！
    protected final boolean compareAndSetState(int expect, int update) {
        // See below for intrinsics setup to support this
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

AQS的资源共享方式

/** Marker to indicate a node is waiting in shared mode */
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
static final Node EXCLUSIVE = null;

Exclusive-独占，只有一个线程能执行，如ReentrantLock。
Share-共享，多个线程可以同时执行，如Semaphore/CountDownLatch。

AQS的节点状态

static final int CANCELLED =  1;
static final int SIGNAL    = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;

值为0，初始化状态，表示当前节点在sync队列中，等待着获取锁。
CANCELLED，值为1，表示当前的线程被取消；
SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，也就是unpark；
CONDITION，值为-2，表示当前节点在等待condition，也就是在condition队列中；
PROPAGATE，值为-3，表示当前场景下后续的acquireShared能够得以执行；

自定义同步器实现时重点关注什么？

    // 该线程是否正在独占资源。只有用到condition才需要去实现它。
    protected boolean isHeldExclusively() {
        throw new UnsupportedOperationException();
    }

    // 独占方式。尝试获取资源，成功则返回true，失败则返回false。
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

    // 独占方式。尝试释放资源，成功则返回true，失败则返回false。
    protected boolean tryRelease(int arg) {
        throw new UnsupportedOperationException();
    }

    // 共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
    protected int tryAcquireShared(int arg) {
        throw new UnsupportedOperationException();
    }

    // 共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。
    protected boolean tryReleaseShared(int arg) {
        throw new UnsupportedOperationException();
    }

我的自定义锁

public class ZhangWeiLock extends AbstractQueuedSynchronizer {

	/**
	 * 尝试获取锁
	 */
	@Override
	protected boolean tryAcquire(int arg) {
		// cas把state=0变为state=1
		if (compareAndSetState(0, 1)) {
			// 设置执行的线程
			setExclusiveOwnerThread(Thread.currentThread());
			// 返回获取锁成功
			return true;
		}

		// 返回获取锁成功
		return false;
	}

	/**
	 * 释放锁
	 */
	@Override
	protected boolean tryRelease(int arg) {
		// 只有获取锁的线程才能释放锁
		setExclusiveOwnerThread(null);
		setState(0);

		// 返回释放锁成功
		return true;
	}

	public void lock() {
		acquire(1);
	}

	public boolean tryLock() {
		return tryAcquire(1);
	}

	public void unlock() {
		release(1);
	}

	public boolean isLocked() {
		return isHeldExclusively();
	}

}

测试方法

public class TsetZhangWeiLock {

	// 定义数据
	private static int sum = 0;

	// 定义自己的锁
	private static ZhangWeiLock lock = new ZhangWeiLock();

	public static void main(String[] args) throws InterruptedException {

		// 循环3吃，开启三个线程
		for (int i = 0; i < 3; i++) {
			Thread thread = new Thread(() -> {
				// 加锁
				lock.lock();
				try {
					// 不加锁这里执行完小于30000
					for (int j = 0; j < 10000; j++) {
						sum++;
					}
				} finally {
					// 解锁
					lock.unlock();
				}
			});
			thread.start();
		}

		// 等待线程执行完
		Thread.sleep(2000);

		// 打印最后的结果
		System.out.println(sum);
	}
}

测试结果

JDK中是如何自定义同步器的？

一般是通过一个内部类Sync继承 AQS
将同步器所有调用都映射到Sync对应的方法
后续基于源码深入分析各种场景的锁。

AQS的等待唤醒机制内部类：Condition

调用Condition#await方法会释放当前持有的锁，然后阻塞当前线程，同时向Condition队列尾部添加一个节点，所以调用Condition#await方法的时候必须持有锁。
调用Condition#signal方法会将Condition队列的首节点移动到阻塞队列尾部，然后唤醒因调用Condition#await方法而阻塞的线程(唤醒之后这个线程就可以去竞争锁了)，所以调用Condition#signal方法的时候必须持有锁，持有锁的线程唤醒被因调用Condition#await方法而阻塞的线程。

在AbstractQueuedSynchronizer类中的源码位置

    public class ConditionObject implements Condition, java.io.Serializable {

Condition接口的方法

验证等待唤醒：使用ReentrantLock（后续有针对ReentrantLock深入的源码分析）

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TsstWait {

	public static void main(String[] args) {

		Lock lock = new ReentrantLock();
		Condition condition = lock.newCondition();

		new Thread(() -> {
			lock.lock();
			try {
				log.debug(Thread.currentThread().getName() + " 开始处理任务");
				condition.await();
				log.debug(Thread.currentThread().getName() + " 结束处理任务");
			} catch (InterruptedException e) {
				e.printStackTrace();
			} finally {
				lock.unlock();
			}
		}).start();

		new Thread(() -> {
			lock.lock();
			try {
				log.debug(Thread.currentThread().getName() + " 开始处理任务");

				Thread.sleep(2000);
				condition.signal();
				log.debug(Thread.currentThread().getName() + " 结束处理任务");
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				lock.unlock();
			}
		}).start();
	}
}

等待唤醒结果

结束语

获取更多有价值的文章，让我们一起成为架构师！
关注公众号，可以让你对MySQL有非常深入的了解
关注公众号，每天持续高效的了解并发编程！
关注公众号，后续持续高效的了解spring源码！
这个公众号，无广告！！！每日更新！！！

码农公寓

AQS为什么要学？

java层面如何实现管程？

AQS是什么？

AQS的同步等待队列：

AQS的条件等待队列：

AQS必须具备的功能

AQS的内部属性：volatile int state

AQS的资源共享方式

AQS的节点状态

自定义同步器实现时重点关注什么？

我的自定义锁

测试方法

测试结果

JDK中是如何自定义同步器的？

AQS的等待唤醒机制内部类：Condition

在AbstractQueuedSynchronizer类中的源码位置

Condition接口的方法

验证等待唤醒：使用ReentrantLock（后续有针对ReentrantLock深入的源码分析）

等待唤醒结果

结束语

相关文章