当前位置：首页 > news >正文

java并发之AQS

news 2025/7/13 16:41:04

一、简介

AQS，全称：AbstractQueuedSynchronizer，是一个JDK提供的用于构建锁、同步器等线程协作工具类的框架，内部维护FIFO双向队列（双向链表实现）。
AQS重要属性：

// 表示同步状态。它既可以表示独占模式下的锁状态，也可以表示共享模式下的资源数量。通过修改state字段，可以实现多线程的独占或共享模式‌
private volatile int state
// 当前持有独占锁的线程
private transient Thread exclusiveOwnerThread
// 头节点
private transient volatile Node head;
// 尾节点
private transient volatile Node tail;

Node节点重要属性：

// 加入队列的线程
volatile Thread thread;
// 前驱节点
volatile Node prev;
// 后继节点
volatile Node next;
// CANCELLED: 表示线程已经取消了对同步状态的请求。
// SIGNAL: 表示线程需要被唤醒（通常是因为其他线程释放了同步状态）。
// CONDITION: 表示线程正在等待某个条件。
// PROPAGATE: 表示下一次共享状态的释放应该传播到其他线程。
// 0: 初始状态，表示节点没有特定的状态。
volatile int waitStatus;
Node nextWaiter;

AQS 在 ReentrantLock、ReentrantReadWriteLock、Semaphore、CountDownLatch、ThreadPoolExcutor 的 Worker 中都有运用（JDK 1.8），AQS 是这些类的底层原理。

二、实现自定义线程协作工具类

2.1 实现独占锁

重写AQS以下方法

boolean tryAcquire(int arg)
boolean tryRelease(int arg)
boolean isHeldExclusively()

调用AQS以下方法

public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();
}public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;
}

2.2 实现共享锁

重写AQS以下方法

int tryAcquireShared(int arg)
boolean tryReleaseShared(int arg)

调用AQS以下方法

public final void acquireShared(int arg) {if (tryAcquireShared(arg) < 0)doAcquireShared(arg);
}public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;
}public final void acquireSharedInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();if (tryAcquireShared(arg) < 0)doAcquireSharedInterruptibly(arg);
}

2.3 示例

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;public class Test {class MySync extends AbstractQueuedSynchronizer {@Overrideprotected boolean tryAcquire(int arg) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}@Overrideprotected boolean tryRelease(int arg) {setExclusiveOwnerThread(null);setState(0);return true;}@Overrideprotected boolean isHeldExclusively() {return getState() == 1;}public Condition newCondition() {return new ConditionObject();}}class MyLock implements Lock {private MySync sync = new MySync();@Overridepublic void lock() {sync.acquire(1);}@Overridepublic void lockInterruptibly() throws InterruptedException {sync.acquireInterruptibly(1);}@Overridepublic boolean tryLock() {return sync.tryAcquire(1);}@Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {return sync.tryAcquireNanos(1, unit.toNanos(time));}@Overridepublic void unlock() {sync.release(1);}@Overridepublic Condition newCondition() {return sync.newCondition();}}public static void main(String[] args) throws Exception {Test test = new Test();MyLock myLock = test.new MyLock();Thread t1 = new Thread(new Runnable() {@Overridepublic void run() {try {myLock.lock();System.out.println(Thread.currentThread().getName() + "执行开始");Thread.sleep(5000L);System.out.println(Thread.currentThread().getName() + "执行结束");} catch (InterruptedException e) {e.printStackTrace();} finally {myLock.unlock();}}}, "t1");Thread t2 = new Thread(new Runnable() {@Overridepublic void run() {try {myLock.lock();System.out.println(Thread.currentThread().getName() + "执行开始");Thread.sleep(3000L);System.out.println(Thread.currentThread().getName() + "执行结束");} catch (InterruptedException e) {e.printStackTrace();} finally {myLock.unlock();}}}, "t2");Thread t3 = new Thread(new Runnable() {@Overridepublic void run() {try {myLock.lock();System.out.println(Thread.currentThread().getName() + "执行开始");Thread.sleep(1000L);System.out.println(Thread.currentThread().getName() + "执行结束");} catch (InterruptedException e) {e.printStackTrace();} finally {myLock.unlock();}}}, "t3");t1.start();t2.start();t3.start();}
}