Java AQS：并发编程的核心机制

一、引言

在 Java 并发编程中，同步机制是确保多线程环境下数据一致性和线程安全的关键。Java 中的 AbstractQueuedSynchronizer（AQS）是一个非常重要的同步框架，它为构建锁和同步器提供了基础架构。本文将深入探讨 Java AQS 的原理、实现方式、应用场景以及实际示例，帮助读者更好地理解和运用这一强大的并发编程工具。

二、AQS 的基本概念

（一）什么是 AQS

AQS 即 AbstractQueuedSynchronizer，是 Java 并发包中的一个抽象类，它提供了一种实现同步器的框架。同步器可以用于实现锁、信号量、条件变量等并发工具。AQS 维护了一个等待队列，用于存储等待获取同步状态的线程。当一个线程尝试获取同步状态失败时，它会被加入到等待队列中，等待其他线程释放同步状态。当一个线程释放同步状态时，AQS 会唤醒等待队列中的一个线程，使其有机会获取同步状态。

（二）AQS 的核心组成部分

1. 同步状态（state）：AQS 使用一个整数变量来表示同步状态。不同的同步器可以根据自己的需求对同步状态进行不同的解释。例如，对于一个独占锁，同步状态可以表示锁是否被占用；对于一个信号量，同步状态可以表示剩余的许可证数量。
2. 等待队列（CLH 队列）：AQS 维护了一个基于链表的等待队列，用于存储等待获取同步状态的线程。等待队列是一个双向链表，每个节点表示一个等待线程。当一个线程尝试获取同步状态失败时，它会被封装成一个节点并加入到等待队列的尾部。当一个线程释放同步状态时，AQS 会从等待队列的头部唤醒一个等待线程。
3. 独占模式和共享模式：AQS 支持两种获取同步状态的模式，即独占模式和共享模式。在独占模式下，每次只有一个线程可以获取同步状态；在共享模式下，多个线程可以同时获取同步状态。不同的同步器可以根据自己的需求选择使用独占模式或共享模式。

三、AQS 的工作原理

（一）独占模式下的获取和释放同步状态

1. 获取同步状态
   • 在独占模式下，当一个线程调用同步器的 acquire 方法尝试获取同步状态时，AQS 会首先检查同步状态是否为 0。如果同步状态为 0，表示当前没有线程占用同步状态，该线程可以获取同步状态，并将同步状态设置为 1。如果同步状态不为 0，表示当前有线程占用同步状态，该线程会被加入到等待队列中，进入阻塞状态。
   • 等待队列中的线程会不断地检查同步状态是否变为 0。当一个线程释放同步状态时，AQS 会唤醒等待队列中的一个线程，使其有机会获取同步状态。被唤醒的线程会再次尝试获取同步状态，如果成功，则继续执行；如果失败，则继续在等待队列中等待。
2. 释放同步状态
   • 在独占模式下，当一个线程调用同步器的 release 方法释放同步状态时，AQS 会将同步状态减 1。如果同步状态变为 0，表示没有线程占用同步状态，此时 AQS 会检查等待队列中是否有等待线程。如果有等待线程，AQS 会唤醒等待队列中的一个线程，使其有机会获取同步状态。

（二）共享模式下的获取和释放同步状态

1. 获取同步状态
   • 在共享模式下，当一个线程调用同步器的 acquireShared 方法尝试获取同步状态时，AQS 会首先检查同步状态是否大于 0。如果同步状态大于 0，表示当前有剩余的同步资源，该线程可以获取同步资源，并将同步状态减 1。如果同步状态不大于 0，表示当前没有剩余的同步资源，该线程会被加入到等待队列中，进入阻塞状态。
   • 等待队列中的线程会不断地检查同步状态是否大于 0。当一个线程释放同步资源时，AQS 会将同步状态加 1。如果同步状态变为大于 0，表示有新的同步资源可用，此时 AQS 会检查等待队列中是否有等待线程。如果有等待线程，AQS 会唤醒等待队列中的一个或多个线程，使其有机会获取同步资源。
2. 释放同步状态
   • 在共享模式下，当一个线程调用同步器的 releaseShared 方法释放同步资源时，AQS 会将同步状态加 1。如果同步状态变为大于 0，表示有新的同步资源可用，此时 AQS 会检查等待队列中是否有等待线程。如果有等待线程，AQS 会唤醒等待队列中的一个或多个线程，使其有机会获取同步资源。

（三）等待队列的实现原理

1. CLH 队列的结构
   • AQS 的等待队列是一个基于链表的队列，称为 CLH（Craig, Landin, and Hagersten）队列。CLH 队列是一种自旋锁队列，每个节点表示一个等待线程。节点包含一个线程引用和一个指向前一个节点的指针。队列的头部节点表示当前正在占用同步状态的线程，尾部节点表示最后一个加入等待队列的线程。
2. 线程加入等待队列的过程
   • 当一个线程尝试获取同步状态失败时，它会被封装成一个节点并加入到等待队列的尾部。具体过程如下：
     • 首先，创建一个节点，并将其线程引用设置为当前线程。
     • 然后，通过 CAS 操作将节点加入到等待队列的尾部。如果 CAS 操作失败，表示有其他线程正在加入等待队列，此时当前线程会自旋等待，直到 CAS 操作成功。
     • 最后，将当前线程的状态设置为等待状态，并将其前驱节点的状态设置为 SIGNAL，表示当前线程需要被唤醒。
3. 线程从等待队列中唤醒的过程
   • 当一个线程释放同步状态时，AQS 会唤醒等待队列中的一个线程。具体过程如下：
     • 首先，检查等待队列的头部节点是否为空。如果为空，表示没有等待线程，直接返回。
     • 如果头部节点不为空，检查头部节点的线程是否为当前线程。如果是，表示当前线程正在占用同步状态，不需要唤醒其他线程，直接返回。
     • 如果头部节点的线程不是当前线程，检查头部节点的状态是否为 SIGNAL。如果是，表示头部节点的线程需要被唤醒。此时，通过 CAS 操作将头部节点的状态设置为 0，表示该节点已经被处理。然后，获取头部节点的线程引用，并唤醒该线程。
     • 如果头部节点的状态不是 SIGNAL，表示头部节点的线程不需要被唤醒。此时，需要找到等待队列中第一个需要被唤醒的节点，并将其状态设置为 SIGNAL。然后，唤醒该节点的线程。

四、AQS 的应用场景

（一）实现锁

1. 独占锁
   • ReentrantLock 是 Java 中最常用的独占锁实现之一，它内部使用了 AQS 来实现锁的功能。ReentrantLock 支持公平锁和非公平锁两种模式。在公平锁模式下，等待队列中的线程按照先来先服务的原则获取锁；在非公平锁模式下，等待队列中的线程和新加入的线程都有机会竞争锁。
   • 以下是一个使用 ReentrantLock 的示例：

import java.util.concurrent.locks.ReentrantLock;public class ReentrantLockExample {public static void main(String[] args) {ReentrantLock lock = new ReentrantLock();try {lock.lock();System.out.println("Thread " + Thread.currentThread().getName() + " acquired the lock.");} finally {lock.unlock();System.out.println("Thread " + Thread.currentThread().getName() + " released the lock.");}}
}

1. 读写锁
   • ReadWriteLock 是 Java 中用于实现读写分离的锁机制，它内部使用了 AQS 来实现锁的功能。ReadWriteLock 支持多个线程同时读取共享资源，但在写入共享资源时，必须独占访问。
   • 以下是一个使用 ReadWriteLock 的示例：

import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;public class ReadWriteLockExample {public static void main(String[] args) {ReadWriteLock lock = new ReentrantReadWriteLock();try {// 获取读锁lock.readLock().lock();System.out.println("Thread " + Thread.currentThread().getName() + " acquired the read lock.");// 模拟读取共享资源Thread.sleep(1000);// 释放读锁lock.readLock().unlock();System.out.println("Thread " + Thread.currentThread().getName() + " released the read lock.");} catch (InterruptedException e) {e.printStackTrace();}try {// 获取写锁lock.writeLock().lock();System.out.println("Thread " + Thread.currentThread().getName() + " acquired the write lock.");// 模拟写入共享资源Thread.sleep(1000);// 释放写锁lock.writeLock().unlock();System.out.println("Thread " + Thread.currentThread().getName() + " released the write lock.");} catch (InterruptedException e) {e.printStackTrace();}}
}

（二）实现信号量

1. Semaphore 简介
   • Semaphore 是 Java 中用于控制同时访问某个资源的线程数量的同步工具，它内部使用了 AQS 来实现信号量的功能。Semaphore 可以用于实现资源池、流量控制等场景。
2. 示例代码
   • 以下是一个使用 Semaphore 的示例：

import java.util.concurrent.Semaphore;public class SemaphoreExample {public static void main(String[] args) {Semaphore semaphore = new Semaphore(3); // 最多允许 3 个线程同时访问资源for (int i = 0; i < 10; i++) {new Thread(() -> {try {semaphore.acquire();System.out.println("Thread " + Thread.currentThread().getName() + " acquired a permit.");// 模拟访问资源Thread.sleep(1000);semaphore.release();System.out.println("Thread " + Thread.currentThread().getName() + " released a permit.");} catch (InterruptedException e) {e.printStackTrace();}}).start();}}
}

（三）实现条件变量

1. Condition 简介
   • Condition 是 Java 中用于实现线程间等待 / 通知机制的同步工具，它内部使用了 AQS 来实现条件变量的功能。Condition 可以与锁配合使用，实现线程间的等待和通知。
2. 示例代码
   • 以下是一个使用 Condition 的示例：

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;public class ConditionExample {public static void main(String[] args) {ReentrantLock lock = new ReentrantLock();Condition condition = lock.newCondition();new Thread(() -> {lock.lock();try {System.out.println("Thread 1 is waiting.");condition.await();System.out.println("Thread 1 is notified.");} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}}).start();new Thread(() -> {lock.lock();try {Thread.sleep(2000);System.out.println("Thread 2 is notifying.");condition.signal();} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}}).start();}
}

五、AQS 的高级特性

（一）可中断的获取同步状态

1. 独占模式下的可中断获取
   • 在独占模式下，线程可以通过调用同步器的 acquireInterruptibly 方法来尝试获取同步状态。这个方法与 acquire 方法类似，但是它可以被中断。如果线程在等待获取同步状态的过程中被中断，它会抛出 InterruptedException 异常，并将中断状态设置为 true。
2. 共享模式下的可中断获取
   • 在共享模式下，线程可以通过调用同步器的 acquireSharedInterruptibly 方法来尝试获取同步资源。这个方法与 acquireShared 方法类似，但是它可以被中断。如果线程在等待获取同步资源的过程中被中断，它会抛出 InterruptedException 异常，并将中断状态设置为 true。

（二）超时获取同步状态

1. 独占模式下的超时获取
   • 在独占模式下，线程可以通过调用同步器的 tryAcquireNanos 方法来尝试在指定的时间内获取同步状态。这个方法与 acquire 方法类似，但是它会在指定的时间内等待获取同步状态。如果在指定的时间内获取到了同步状态，它会返回 true；如果在指定的时间内没有获取到同步状态，它会返回 false。
2. 共享模式下的超时获取
   • 在共享模式下，线程可以通过调用同步器的 tryAcquireSharedNanos 方法来尝试在指定的时间内获取同步资源。这个方法与 acquireShared 方法类似，但是它会在指定的时间内等待获取同步资源。如果在指定的时间内获取到了同步资源，它会返回一个大于等于 0 的值，表示获取到的同步资源数量；如果在指定的时间内没有获取到同步资源，它会返回一个小于 0 的值。

（三）公平性与非公平性

1. 公平锁与非公平锁的区别
   • 在 AQS 中，同步器可以选择实现公平锁或非公平锁。公平锁是指在等待队列中的线程按照先来先服务的原则获取同步状态；非公平锁是指在等待队列中的线程和新加入的线程都有机会竞争锁。
   • 公平锁可以保证等待时间最长的线程优先获取同步状态，但是它会导致性能下降，因为在每次获取同步状态时都需要检查等待队列中是否有等待时间更长的线程。非公平锁可以提高性能，但是它不能保证等待时间最长的线程优先获取同步状态。
2. 如何选择公平锁或非公平锁
   • 在选择公平锁或非公平锁时，需要考虑以下因素：
     • 性能要求：如果对性能要求较高，可以选择非公平锁；如果对公平性要求较高，可以选择公平锁。
     • 等待时间：如果等待时间较长，可以选择公平锁，以保证等待时间最长的线程优先获取同步状态；如果等待时间较短，可以选择非公平锁，以提高性能。
     • 线程数量：如果线程数量较少，可以选择公平锁，以保证公平性；如果线程数量较多，可以选择非公平锁，以提高性能。

六、AQS 的内部实现细节

（一）同步状态的管理

1. 同步状态的表示
   • AQS 使用一个整数变量来表示同步状态。这个整数变量可以表示不同的同步状态，具体的含义由同步器的实现者来定义。例如，对于一个独占锁，同步状态可以表示锁是否被占用；对于一个信号量，同步状态可以表示剩余的许可证数量。
2. 同步状态的修改
   • AQS 提供了一些方法来修改同步状态，如 compareAndSetState、getState 和 setState 方法。这些方法都是基于 CAS（Compare and Swap）操作实现的，保证了同步状态的原子性修改。
   • compareAndSetState 方法用于比较并设置同步状态。它会先比较当前同步状态与预期值是否相等，如果相等，则将同步状态设置为新值，并返回 true；如果不相等，则返回 false。
   • getState 方法用于获取当前同步状态的值。
   • setState 方法用于设置当前同步状态的值。

（二）等待队列的管理

1. 等待队列的节点结构
   • AQS 的等待队列是一个基于链表的队列，每个节点表示一个等待线程。节点包含一个线程引用、一个指向前一个节点的指针和一个表示节点状态的变量。节点状态可以是 SIGNAL、CANCELLED 或 CONDITION 等。
2. 等待队列的操作
   • AQS 提供了一些方法来操作等待队列，如 enq、addWaiter 和 unparkSuccessor 方法。这些方法都是基于 CAS 操作实现的，保证了等待队列的线程安全。
   • enq 方法用于将一个节点加入到等待队列的尾部。它会先创建一个空的节点，如果等待队列的头部为空，则将该节点设置为等待队列的头部和尾部；如果等待队列的头部不为空，则通过 CAS 操作将该节点加入到等待队列的尾部。
   • addWaiter 方法用于将当前线程封装成一个节点并加入到等待队列中。它会先检查当前线程是否已经在等待队列中，如果是，则返回该节点；如果不是，则创建一个新的节点，并将其加入到等待队列中。
   • unparkSuccessor 方法用于唤醒等待队列中的下一个节点。它会先检查等待队列的头部节点是否为空，如果为空，则直接返回；如果头部节点不为空，则检查头部节点的状态是否为 SIGNAL。如果是，则将头部节点的状态设置为 0，并唤醒头部节点的线程；如果不是，则找到等待队列中第一个需要被唤醒的节点，并将其状态设置为 SIGNAL，然后唤醒该节点的线程。

七、AQS 的性能优化

（一）减少锁竞争

1. • 优化同步器的实现

     • 在实现同步器时，可以通过优化同步状态的管理和等待队列的操作来减少锁竞争。例如，可以使用更高效的算法来判断同步状态是否可用，或者使用更优化的等待队列结构来减少线程的等待时间。
     • 对于独占锁，可以考虑使用自旋锁来减少线程的阻塞和唤醒开销。自旋锁是一种在获取锁时不断循环尝试的锁机制，它适用于锁被占用时间较短的场景。当线程尝试获取锁时，如果锁已经被占用，线程会在一个循环中不断地检查锁是否可用，而不是立即进入阻塞状态。如果锁在短时间内被释放，线程可以快速地获取锁，避免了线程的阻塞和唤醒开销。
     • 对于共享锁，可以考虑使用分段锁来减少锁竞争。分段锁是一种将数据分成多个段，每个段使用一个独立的锁进行保护的锁机制。当多个线程同时访问不同段的数据时，它们可以并行地获取锁，而不会相互阻塞。只有当多个线程同时访问同一段数据时，才会发生锁竞争。

   • 避免不必要的同步

     • 在编写代码时，应该尽量避免不必要的同步操作。例如，可以使用局部变量代替共享变量，或者使用不可变对象来避免对共享数据的修改。
     • 如果一个方法不需要对共享数据进行修改，那么可以将该方法声明为非同步方法，以提高性能。
     • 另外，可以使用线程安全的集合类来代替传统的集合类，以避免手动进行同步操作。Java 提供了一些线程安全的集合类，如 ConcurrentHashMap、CopyOnWriteArrayList 等，它们在多线程环境下具有更好的性能。

（二）优化等待队列的操作

 

减少节点的创建和销毁

• AQS 的等待队列是一个基于链表的队列，每个节点表示一个等待线程。在高并发的情况下，节点的创建和销毁可能会成为性能瓶颈。为了减少节点的创建和销毁，可以考虑使用对象池来管理节点对象。
• 对象池是一种预先创建一定数量的对象，并在需要时从池中获取对象，使用完毕后将对象归还到池中以供下次使用的技术。通过使用对象池，可以减少对象的创建和销毁开销，提高性能。
• 在实现 AQS 的等待队列时，可以使用对象池来管理节点对象。当一个线程需要加入等待队列时，可以从对象池中获取一个节点对象，并将其加入到等待队列中。当一个线程从等待队列中被唤醒时，可以将其对应的节点对象归还到对象池中，以供下次使用。

优化节点的状态管理

• AQS 的等待队列中的节点有不同的状态，如 SIGNAL、CANCELLED 和 CONDITION 等。在高并发的情况下，节点状态的管理可能会成为性能瓶颈。为了优化节点的状态管理，可以考虑使用位运算来表示节点状态。
• 位运算是一种高效的运算方式，可以在不使用额外的内存空间的情况下表示多个状态。通过使用位运算，可以减少节点状态的存储空间，提高性能。
• 在实现 AQS 的等待队列时，可以使用位运算来表示节点状态。例如，可以使用一个整数变量来表示节点状态，其中不同的位表示不同的状态。当需要检查节点状态时，可以使用位运算来快速判断节点的状态。

 

（三）使用批量操作

 

批量获取和释放同步状态

• 在某些情况下，可以使用批量操作来提高性能。例如，如果多个线程需要同时获取或释放同步状态，可以考虑使用批量操作来减少锁竞争和等待时间。
• 在实现同步器时，可以提供批量获取和释放同步状态的方法。这些方法可以一次性处理多个线程的请求，减少锁竞争和等待时间。
• 例如，可以实现一个批量获取锁的方法，该方法可以接受一个线程数组作为参数，并一次性为所有线程获取锁。当所有线程都获取到锁后，方法才返回。这样可以避免多个线程逐个获取锁时的锁竞争和等待时间。

批量唤醒等待线程

• 同样地，可以实现批量唤醒等待线程的方法。当一个线程释放同步状态时，可以一次性唤醒多个等待线程，而不是逐个唤醒。这样可以减少唤醒等待线程的开销，提高性能。
• 在实现 AQS 的等待队列时，可以提供批量唤醒等待线程的方法。当一个线程释放同步状态时，可以检查等待队列中是否有多个等待线程需要被唤醒。如果有，可以一次性唤醒多个等待线程，而不是逐个唤醒。

 

八、总结

 

Java 的 AbstractQueuedSynchronizer（AQS）是一个强大的并发编程工具，它为构建锁和同步器提供了基础架构。通过理解 AQS 的工作原理、应用场景、高级特性以及内部实现细节，开发者可以更好地利用它来实现高效的并发程序。同时，通过性能优化措施，可以进一步提高 AQS 的性能，满足高并发场景下的需求。在实际应用中，开发者应根据具体的业务需求和性能要求，选择合适的同步机制，并合理地使用 AQS 来实现线程安全和并发控制。