AQS 详解 | JavaGuide

Semaphore 信号量

synchronized  和  ReentrantLock  都是一次只允许一个线程访问某个资源，而Semaphore(信号量)可以用来控制同时访问特定资源的线程数量。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// 初始共享资源数量
final Semaphore semaphore = new Semaphore(5);
// 获取1个许可
semaphore.acquire();
// 释放1个许可
semaphore.release();


同一时刻 只有 5 个线程能获取到共享资源，其他线程都会阻塞，只有获取到共享资源的线程才能执行。
等到有线程释放了共享资源，其他阻塞的线程才能获取到。

当初始的资源个数为 1 ，Semaphore 退化为排他锁

• Semaphore 公平模式
  调用  acquire()  方法的顺序就是获取许可证的顺序，遵循 FIFO；
• Semaphore 抢占模式

Semaphore 两个构造函数如下：

1
2
3
4
5
6
7
8

//默认为不公平
public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

原理

Semaphore 是共享锁的一种实现，它默认构造 AQS 的 state 值为 permits， permits 为许可证的数量，只有拿到许可证的线程才能执行。

以无参 acquire 方法为例，调用semaphore.acquire() ，线程尝试获取许可证。

如果 state > 0 ，则获取成功，此时会尝试使用 CAS 操作去修改 state 的值 state=state-1

如果 state <= 0 ，则许可证数量不足，获取失败，会创建一个 Node 节点加入等待队列，挂起当前线程。

以无参 release 方法为例，调用semaphore.release()，线程尝试释放许可证，并使用 CAS 操作去修改 state 的值 state=state+1。释放许可证成功之后，同时会唤醒等待队列中的一个线程。被唤醒的线程会重新尝试修改 state 的值 state=state-1 ，如果 state > 0 则获取许可证成功，否则重新进入等待队列（重新排队而不是在这个位置继续等），继续挂起线程。

CountDownLatch 倒计时器

CountDownLatch 允许  count  个线程阻塞在一个地方，直至所有线程的任务都执行完毕。

CountDownLatch  是一次性的，计数器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当  CountDownLatch  使用完毕后不能再次被使用。

原理

• CountDownLatch  也是共享锁的一种实现，它默认构造 AQS 的  state  值为  count。
• 当线程调用 countDown() 时，其实在 releaseShared 方法中调用 tryReleaseShared 方法以 CAS 的操作来减少 state，直至 state 为 0 ，此时表示所有的线程都调用了 countDown 方法，那么在 CountDownLatch 上等待的线程就会被唤醒并继续执行。
• 以无参 await方法为例，当调用 await() 的时候，如果 state 不为 0，那就证明任务还没有执行完毕，await() 就会一直阻塞，也就是说 await() 之后的语句不会被执行（main 线程被加入到 AQS 的 CLH 队列中了）。然后，CountDownLatch 会自旋 CAS 判断 state == 0，如果 state == 0 的话，就会释放所有等待的线程，await() 方法之后的语句得到执行。

1
2
3
4

public void countDown() {
    // Sync 是 CountDownLatch 的内部类 , 继承了 AbstractQueuedSynchronizer
    sync.releaseShared(1);
}

CountDownLatch用法

• 某一线程在开始运行前等待 n 个线程执行完毕 : 将 CountDownLatch 的计数器初始化为 n （new CountDownLatch(n)），每当一个任务线程执行完毕，就将计数器减 1 （countdownlatch.countDown()），当计数器的值变为 0 时，在 CountDownLatch 上 await() 的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行（CompletableFuture 也可以做异步任务编排）。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

import java.util.concurrent.CountDownLatch;

public class ServiceStartup {
    public static void main(String[] args) throws InterruptedException {
        // 假设有三个组件需要加载
        CountDownLatch latch = new CountDownLatch(3);

        // 启动多个线程模拟组件加载
        new Thread(() -> {
            loadComponent("Component 1", latch);
        }).start();

        new Thread(() -> {
            loadComponent("Component 2", latch);
        }).start();

        new Thread(() -> {
            loadComponent("Component 3", latch);
        }).start();

        // 主线程等待所有组件加载完毕
        latch.await();

        // 所有组件加载完成后继续执行
        System.out.println("All components loaded. Service is ready to start.");
    }

    private static void loadComponent(String componentName, CountDownLatch latch) {
        try {
            System.out.println(componentName + " is loading...");
            Thread.sleep(1000); // 模拟组件加载时间
            latch.countDown();
            System.out.println(componentName + " loaded.");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

• 实现多个线程开始执行任务的最大并行性：注意是并行性，不是并发，强调的是多个线程在**某一时刻同时开始执行，**类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的 CountDownLatch 对象，主线程将其计数器初始化为 1 （new CountDownLatch(1)），多个线程在开始执行任务前首先 coundownlatch.await()，当主线程调用 countDown() 时，计数器变为 0，多个线程同时被唤醒。

CyclicBarrier 循环栅栏

• CyclicBarrier 和 CountDownLatch 非常类似，它也可以实现线程间的技术等待，但功能比 CountDownLatch 更加复杂和强大。
• CountDownLatch 的实现是基于 AQS 的，而 CycliBarrier 是基于ReentrantLock和 Condition 的。
• CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。作用是：让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活。

原理

private final int parties; 每次拦截的线程数   private int count;  计数器

CyclicBarrier 内部通过一个 count 变量作为计数器，count 的初始值为 parties 属性的初始化值，每当一个线程到了栅栏这里了，那么就将计数器减 1。如果 count 值为 0 了，表示这是这一代最后一个线程到达栅栏，就尝试执行我们构造方法中输入的任务。