深入理解高并发编程 - AQS 中的 CountDownLatch、Semaphore 与 CyclicBarrier

Bertha 。 2023-10-15 09:19 121阅读 0赞

## AQS ##

AQS（AbstractQueuedSynchronizer）是 Java 中用于实现同步器的一个抽象基类。它提供了构建各种同步工具的基础框架，如锁、信号量、倒计时器等。AQS 主要通过一个先进先出（FIFO）的等待队列来管理线程的竞争和等待状态。

AQS 的核心思想是基于状态的控制和等待/通知机制。每个 AQS 子类都可以定义一个状态（state），并提供了一些基本方法来操作这个状态，例如获取锁、释放锁等。基于这些方法，AQS 能够实现线程的排他性访问以及线程之间的协调。

AQS 主要有两种同步模式：独占模式（Exclusive Mode）和共享模式（Shared Mode）。

独占模式（Exclusive Mode）：在独占模式下，同一时刻只能有一个线程持有锁。ReentrantLock 就是一个基于独占模式的同步工具。
    
    共享模式（Shared Mode）：在共享模式下，多个线程可以同时获取锁。Semaphore 和 CountDownLatch 等就是基于共享模式的同步工具。

AQS 内部维护了一个等待队列，等待队列中的线程按照先进先出的顺序等待获取同步状态。当一个线程获取不到同步状态时，它会被包装成一个节点（Node）并加入到等待队列中，然后进入阻塞状态。当同步状态可用时，AQS 会根据特定的策略从等待队列中选取一个线程来获取同步状态。

在自定义同步器时，通常需要重写 AQS 的一些方法，主要包括：

getState 和 setState：用于获取和设置同步状态。
    tryAcquire 和 tryRelease：用于独占模式下的状态获取和释放。
    tryAcquireShared 和 tryReleaseShared：用于共享模式下的状态获取和释放。

AQS 提供了一些工具方法和钩子方法，用于实现具体的同步逻辑和管理等待队列。

需要注意的是，直接使用 AQS 需要深入理解其原理，因为涉及线程状态管理和等待队列的细节。不过，大部分情况下，开发者会使用基于 AQS 的高级同步工具，如 ReentrantLock、Semaphore、CountDownLatch 等，而不需要直接与 AQS 打交道。

## CountDownLatch ##

CountDownLatch 是 Java 提供的一个同步工具类，它允许一个或多个线程等待一组操作完成。它的主要思想是：一个或多个线程等待其他线程完成一定数量的操作，然后再继续执行。

CountDownLatch 主要有两个方法：await() 和 countDown()。

await(): 当调用这个方法的线程调用时，它会阻塞，直到计数器减到零。换句话说，线程会等待直到所有需要等待的操作都完成。
    countDown(): 每次调用这个方法会使计数器减一。一般情况下，当某个操作完成后，应该调用这个方法来通知 CountDownLatch 计数器减少。

以下是一个简单的示例，说明如何使用 CountDownLatch：

import java.util.concurrent.CountDownLatch;
    
    public class CountDownLatchExample {
        public static void main(String[] args) throws InterruptedException {
            int workerCount = 3;
            CountDownLatch latch = new CountDownLatch(workerCount);
    
            for (int i = 0; i < workerCount; i++) {
                new Worker(latch).start();
            }
    
            latch.await(); // 等待所有工作线程完成
            System.out.println("All workers have finished their tasks.");
        }
    
        static class Worker extends Thread {
            private final CountDownLatch latch;
    
            Worker(CountDownLatch latch) {
                this.latch = latch;
            }
    
            @Override
            public void run() {
                System.out.println("Worker " + getId() + " is working.");
                // 模拟工作
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("Worker " + getId() + " has finished its task.");
                latch.countDown(); // 通知CountDownLatch计数减少
            }
        }
    }

在上面的示例中，有三个工作线程在执行任务，每个工作线程完成任务后会调用 countDown() 方法通知 CountDownLatch，最终当所有工作线程都完成后，主线程的 await() 被解除阻塞，继续执行。

CountDownLatch 在并发编程中常用于等待多个线程完成某些操作，然后再继续执行后续任务。

## Semaphore ##

Semaphore 是 Java 提供的另一个同步工具类，用于控制并发访问的数量。它可以用来限制同时访问某个资源或执行某个操作的线程数量。

Semaphore 维护了一个许可（permit）计数，代表可同时执行的线程数量。当线程要执行受保护的操作时，需要从 Semaphore 获取许可。如果许可数大于零，线程将获得许可并执行操作；如果许可数为零，线程将阻塞，直到有其他线程释放许可。

Semaphore 主要有两个构造方法：

Semaphore(int permits): 初始化一个指定数量的许可。
    Semaphore(int permits, boolean fair): 可以指定是否使用公平策略（FIFO）来分配许可。

主要的方法包括：

void acquire(): 获取一个许可。如果没有许可可用，线程将阻塞。
    void release(): 释放一个许可，使许可数加一。如果有其他线程在等待许可，将会有一个线程被唤醒。
    int availablePermits(): 返回当前可用的许可数量。
    int getQueueLength(): 返回等待获取许可的线程数。

以下是一个示例，展示如何使用 Semaphore 控制并发访问：

import java.util.concurrent.Semaphore;
    
    public class SemaphoreExample {
        public static void main(String[] args) {
            int numWorkers = 5;
            Semaphore semaphore = new Semaphore(2); // 允许同时执行的线程数量为2
    
            for (int i = 0; i < numWorkers; i++) {
                new Worker(i, semaphore).start();
            }
        }
    
        static class Worker extends Thread {
            private final int id;
            private final Semaphore semaphore;
    
            Worker(int id, Semaphore semaphore) {
                this.id = id;
                this.semaphore = semaphore;
            }
    
            @Override
            public void run() {
                try {
                    System.out.println("Worker " + id + " is waiting.");
                    semaphore.acquire(); // 获取许可
                    System.out.println("Worker " + id + " has acquired the permit and is working.");
                    // 模拟工作
                    Thread.sleep(2000);
                    System.out.println("Worker " + id + " has finished working.");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    semaphore.release(); // 释放许可
                    System.out.println("Worker " + id + " has released the permit.");
                }
            }
        }
    }

在上述示例中，有5个工作线程，但是只允许同时有2个工作线程获得许可并执行任务。当有多于2个线程时，其他线程会在 acquire() 处阻塞，直到有线程释放许可。这可以用来控制同时并发执行的线程数量，以防止资源过度竞争。

## CyclicBarrier ##

CyclicBarrier 是 Java 提供的另一个同步工具类，用于在多个线程之间进行同步等待，直到一组线程都达到某个共同点之后，它们才会继续执行。

与 CountDownLatch 不同，CyclicBarrier 可以在多次使用时被重置，因此可以循环使用。它在多个线程之间构建了一个屏障，当所有线程都到达这个屏障后，屏障会打开，线程可以继续执行。

CyclicBarrier 的构造方法可以指定等待的线程数量，以及当所有线程到达时要执行的动作（可选）。

主要的方法包括：

await(): 当线程调用这个方法时，它会等待其他线程都到达屏障。一旦所有线程都到达，屏障会打开，线程可以继续执行。此方法还可以指定等待的超时时间。

以下是一个简单的示例，展示了如何使用 CyclicBarrier：

import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    public class CyclicBarrierExample {
        public static void main(String[] args) {
            int numThreads = 3;
            Runnable barrierAction = () -> System.out.println("All threads have reached the barrier.");
            CyclicBarrier barrier = new CyclicBarrier(numThreads, barrierAction);
    
            for (int i = 0; i < numThreads; i++) {
                new Worker(barrier).start();
            }
        }
    
        static class Worker extends Thread {
            private final CyclicBarrier barrier;
    
            Worker(CyclicBarrier barrier) {
                this.barrier = barrier;
            }
    
            @Override
            public void run() {
                try {
                    System.out.println("Worker is doing some work.");
                    Thread.sleep(1000); // 模拟工作
                    System.out.println("Worker has finished its work and is waiting at the barrier.");
                    barrier.await(); // 等待其他线程到达屏障
                    System.out.println("Worker has passed the barrier and continues its work.");
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }
    }

在上述示例中，有3个工作线程，它们在到达屏障前都会进行一些工作。一旦所有线程都到达屏障，屏障打开，所有线程都可以继续执行后续工作。

CyclicBarrier 在某些情况下很有用，例如需要多个线程协调在某个点上同步执行，或者需要在多个线程之间实现阶段性的同步。

## 区别 ##

下面是比较 `CountDownLatch`、`Semaphore` 和 `CyclicBarrier` 这三个 Java 同步工具的区别：

<table> 
 <thead> 
  <tr> 
   <th>特性</th> 
   <th><code>CountDownLatch</code></th> 
   <th><code>Semaphore</code></th> 
   <th><code>CyclicBarrier</code></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>目的</td> 
   <td>等待一组线程完成一定数量的操作，再继续执行</td> 
   <td>控制同时访问某个资源的线程数量</td> 
   <td>多个线程在到达屏障后继续执行</td> 
  </tr> 
  <tr> 
   <td>可重用性</td> 
   <td>不能重用（计数减至零后不能重置）</td> 
   <td>可重用（可以通过 <code>release()</code> 重置许可数）</td> 
   <td>可重用（在每次使用前通过构造函数重置）</td> 
  </tr> 
  <tr> 
   <td>计数方式</td> 
   <td>递减计数</td> 
   <td>许可计数</td> 
   <td>计数增加至指定数量</td> 
  </tr> 
  <tr> 
   <td>超时等待</td> 
   <td>不支持</td> 
   <td>支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>动作</td> 
   <td>无</td> 
   <td>无</td> 
   <td>可以指定在屏障打开时执行的动作</td> 
  </tr> 
  <tr> 
   <td>线程间关系</td> 
   <td>独立等待，无交互</td> 
   <td>无交互，许可的获取和释放不影响其他线程</td> 
   <td>互相等待，屏障开放后线程继续执行</td> 
  </tr> 
  <tr> 
   <td>示例类</td> 
   <td><code>java.util.concurrent.CountDownLatch</code></td> 
   <td><code>java.util.concurrent.Semaphore</code></td> 
   <td><code>java.util.concurrent.CyclicBarrier</code></td> 
  </tr> 
  <tr> 
   <td>使用场景</td> 
   <td>等待多个线程完成某个共同操作后再继续执行</td> 
   <td>控制资源访问数量</td> 
   <td>多线程阶段性同步</td> 
  </tr> 
 </tbody> 
</table>

这个表格列出了这三个同步工具的一些主要区别和特性。需要根据具体的场景和需求来选择使用哪个工具。