Java AQS框架解析 - 元素码农

发布时间: 2025-03-22 09:34

↑

# Java AQS框架解析

AQS（AbstractQueuedSynchronizer）是Java并发包中的核心框架，它提供了一个基于FIFO队列实现的阻塞锁和相关的同步器。本文将详细介绍AQS的实现原理、核心组件以及实践应用。

## AQS基础概念

### 1. 什么是AQS

AQS是一个用于构建锁和同步器的框架，它使用一个int成员变量表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作。

```java
public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
    // 同步状态
    private volatile int state;
    
    // 获取当前同步状态
    protected final int getState() {
        return state;
    }
    
    // 设置同步状态
    protected final void setState(int newState) {
        state = newState;
    }
    
    // 通过CAS设置同步状态
    protected final boolean compareAndSetState(int expect, int update) {
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }
}
```

### 2. AQS的核心思想

```java
public class AQSDemo {
    private static class Sync extends AbstractQueuedSynchronizer {
        protected boolean tryAcquire(int arg) {
            // 尝试获取资源
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
        
        protected boolean tryRelease(int arg) {
            // 释放资源
            if (getState() == 0) {
                throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }
    }
}
```

核心思想：
1. 资源的获取和释放
2. 等待队列的管理
3. 线程的阻塞和唤醒

## 同步队列

### 1. 队列结构

```java
public class Node {
    // 共享模式
    static final Node SHARED = new Node();
    // 独占模式
    static final Node EXCLUSIVE = null;
    
    // 等待状态
    volatile int waitStatus;
    // 前驱节点
    volatile Node prev;
    // 后继节点
    volatile Node next;
    // 等待的线程
    volatile Thread thread;
    
    // 下一个等待节点
    Node nextWaiter;
}
```

队列特点：
1. 双向链表结构
2. 支持独占和共享模式
3. 包含线程等待状态

### 2. 入队和出队

```java
public class QueueOperationDemo {
    private static final AbstractQueuedSynchronizer sync = new AbstractQueuedSynchronizer() {
        @Override
        protected boolean tryAcquire(int arg) {
            return compareAndSetState(0, 1);
        }
        
        @Override
        protected boolean tryRelease(int arg) {
            setState(0);
            return true;
        }
    };
    
    public static void main(String[] args) {
        // 模拟多个线程竞争资源
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                sync.acquire(1);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                sync.release(1);
            }).start();
        }
    }
}
```

操作流程：
1. 线程获取资源失败后加入队列
2. 线程释放资源后唤醒后继节点
3. 维护队列的FIFO特性

## 独占模式

### 1. 独占式获取

```java
public class ExclusiveLockDemo {
    private static class MyLock {
        private static class Sync extends AbstractQueuedSynchronizer {
            protected boolean tryAcquire(int acquires) {
                if (compareAndSetState(0, 1)) {
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
            
            protected boolean tryRelease(int releases) {
                setState(0);
                setExclusiveOwnerThread(null);
                return true;
            }
        }
        
        private final Sync sync = new Sync();
        
        public void lock() {
            sync.acquire(1);
        }
        
        public void unlock() {
            sync.release(1);
        }
    }
    
    public static void main(String[] args) {
        MyLock lock = new MyLock();
        // 使用自定义锁
        lock.lock();
        try {
            // 临界区代码
        } finally {
            lock.unlock();
        }
    }
}
```

独占式特点：
1. 同一时刻只能有一个线程获取资源
2. 其他线程必须等待
3. 支持重入性质

### 2. 条件变量

```java
public class ConditionDemo {
    private static class BoundedBuffer {
        private final MyLock lock = new MyLock();
        private final Condition notFull = lock.newCondition();
        private final Condition notEmpty = lock.newCondition();
        
        private final Object[] items = new Object[100];
        private int putptr, takeptr, count;
        
        public void put(Object x) throws InterruptedException {
            lock.lock();
            try {
                while (count == items.length) {
                    notFull.await();
                }
                items[putptr] = x;
                if (++putptr == items.length) putptr = 0;
                ++count;
                notEmpty.signal();
            } finally {
                lock.unlock();
            }
        }
        
        public Object take() throws InterruptedException {
            lock.lock();
            try {
                while (count == 0) {
                    notEmpty.await();
                }
                Object x = items[takeptr];
                if (++takeptr == items.length) takeptr = 0;
                --count;
                notFull.signal();
                return x;
            } finally {
                lock.unlock();
            }
        }
    }
}
```

条件变量作用：
1. 线程等待和通知
2. 实现线程间协作
3. 管理线程状态

## 共享模式

### 1. 共享式获取

```java
public class SharedLockDemo {
    private static class Semaphore {
        private static class Sync extends AbstractQueuedSynchronizer {
            Sync(int permits) {
                setState(permits);
            }
            
            protected int tryAcquireShared(int acquires) {
                for (;;) {
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 || compareAndSetState(available, remaining)) {
                        return remaining;
                    }
                }
            }
            
            protected boolean tryReleaseShared(int releases) {
                for (;;) {
                    int current = getState();
                    int next = current + releases;
                    if (compareAndSetState(current, next)) {
                        return true;
                    }
                }
            }
        }
        
        private final Sync sync;
        
        public Semaphore(int permits) {
            sync = new Sync(permits);
        }
        
        public void acquire() {
            sync.acquireShared(1);
        }
        
        public void release() {
            sync.releaseShared(1);
        }
    }
}
```

共享式特点：
1. 多个线程可以同时获取资源
2. 资源数量可以动态变化
3. 支持公平和非公平模式

### 2. 实现示例

```java
public class SharedModeDemo {
    private static class CountDownLatch {
        private static class Sync extends AbstractQueuedSynchronizer {
            Sync(int count) {
                setState(count);
            }
            
            protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;
            }
            
            protected boolean tryReleaseShared(int releases) {
                for (;;) {
                    int c = getState();
                    if (c == 0) {
                        return false;
                    }
                    int nextc = c - 1;
                    if (compareAndSetState(c, nextc)) {
                        return nextc == 0;
                    }
                }
            }
        }
        
        private final Sync sync;
        
        public CountDownLatch(int count) {
            sync = new Sync(count);
        }
        
        public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
        
        public void countDown() {
            sync.releaseShared(1);
        }
    }
}
```

实现要点：
1. 状态变量的管理
2. CAS操作保证原子性
3. 线程安全的实现

## 实践应用

### 1. 自定义同步器

```java
public class CustomSynchronizerDemo {
    private static class BooleanLatch {
        private static class Sync extends AbstractQueuedSynchronizer {
            boolean isSignalled() {
                return getState() != 0;
            }
            
            protected int tryAcquireShared(int ignore) {
                return isSignalled() ? 1 : -1;
            }
            
            protected boolean tryReleaseShared(int ignore) {
                setState(1);
                return true;
            }
        }
        
        private final Sync sync = new Sync();
        
        public boolean isSignalled() {
            return sync.isSignalled();
        }
        
        public void signal() {
            sync.releaseShared(1);
        }
        
        public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        BooleanLatch latch = new BooleanLatch();
        
        // 等待线程
        new Thread(() -> {
            try {
                latch.await();
                System.out.println("Latch is signalled");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        
        Thread.sleep(1000);
        
        // 触发信号
        latch.signal();
    }
}
```

实现步骤：
1. 定义同步状态
2. 实现资源获取和释放
3. 封装同步器操作

### 2. 性能优化

```java
public class PerformanceOptimizationDemo {
    private static class OptimizedLock {
        private static class Sync extends AbstractQueuedSynchronizer {
            // 快速路径
            public boolean tryAcquire(int acquires) {
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
            
            // 快速释放
            protected boolean tryRelease(int releases) {
                int c = getState() - releases;
                if (Thread.currentThread() != getExclusiveOwnerThread()) {
                    throw new IllegalMonitorStateException();
                }
                boolean free = false;
                if (c == 0) {
                    free = true;
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }
        }
        
        private final Sync sync = new Sync();
        
        public void lock() {
            if (!sync.tryAcquire(1)) {
                sync.acquire(1); // 降级为完整的获取过程
            }
        }
        
        public void unlock() {
            sync.release(1);
        }
    }
}
```

优化策略：
1. 使用快速路径
2. 减少竞争
3. 优化等待策略

## 总结

通过本文，我们详细了解了AQS框架的核心内容：

1. 基础概念
   - 同步状态管理
   - 等待队列结构
   - 线程调度机制

2. 实现机制
   - 独占模式实现
   - 共享模式实现
   - 条件变量支持

3. 实践应用
   - 自定义同步器
   - 性能优化策略
   - 最佳实践建议

在实际开发中，我们需要：

1. 理解AQS的工作原理
2. 正确使用同步器
3. 注意性能优化
4. 遵循最佳实践

掌握