- 👏作者简介：大家好，我是爱吃芝士的土豆倪，24届校招生Java选手，很高兴认识大家
- 📕系列专栏：Spring源码、JUC源码
- 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
- 🍂博主正在努力完成2023计划中：源码溯源，一探究竟
- 📝联系方式：nhs19990716，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀
文章目录
- Semaphore
- 基本使用
- 限制对共享资源的使用
- semaphore 实现
- Semaphore 原理
- 加锁解锁流程
- CountdownLatch
- 源码
- 应用之同步等待多线程准备完毕
- 应用之同步等待多个远程调用结束
- CyclicBarrier
Semaphore
基本使用
[ˈsɛməˌfɔr] 信号量，用来限制能同时访问共享资源的线程上限。
public static void main(String[] args) {
// 1. 创建 semaphore 对象
Semaphore semaphore = new Semaphore(3);
// 2. 10个线程同时运行
for (int i = 0; i < 10; i++) {
new Thread(() -> {
// 3. 获取许可
try {
semaphore.acquire();
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
log.debug("running...");
sleep(1);
log.debug("end...");
} finally {
// 4. 释放许可
semaphore.release();
}
}).start();
}
}
输出
07:35:15.485 c.TestSemaphore [Thread-2] - running...
07:35:15.485 c.TestSemaphore [Thread-1] - running...
07:35:15.485 c.TestSemaphore [Thread-0] - running...
07:35:16.490 c.TestSemaphore [Thread-2] - end...
07:35:16.490 c.TestSemaphore [Thread-0] - end...
07:35:16.490 c.TestSemaphore [Thread-1] - end...
07:35:16.490 c.TestSemaphore [Thread-3] - running...
07:35:16.490 c.TestSemaphore [Thread-5] - running...
07:35:16.490 c.TestSemaphore [Thread-4] - running...
07:35:17.490 c.TestSemaphore [Thread-5] - end...
07:35:17.490 c.TestSemaphore [Thread-4] - end...
07:35:17.490 c.TestSemaphore [Thread-3] - end...
07:35:17.490 c.TestSemaphore [Thread-6] - running...
07:35:17.490 c.TestSemaphore [Thread-7] - running...
07:35:17.490 c.TestSemaphore [Thread-9] - running...
07:35:18.491 c.TestSemaphore [Thread-6] - end...
07:35:18.491 c.TestSemaphore [Thread-7] - end...
07:35:18.491 c.TestSemaphore [Thread-9] - end...
07:35:18.491 c.TestSemaphore [Thread-8] - running...
07:35:19.492 c.TestSemaphore [Thread-8] - end...
限制对共享资源的使用
semaphore 实现
使用 Semaphore 限流，在访问高峰期时，让请求线程阻塞，高峰期过去再释放许可，当然它只适合限制单机
线程数量，并且仅是限制线程数
用 Semaphore 实现简单连接池，对比『享元模式』下的实现（用wait notify），性能和可读性显然更好
Semaphore 实现简单连接池相对于使用"享元模式"（使用 wait 和 notify）的实现在性能和可读性方面更好的原因主要有以下几点：
简洁的接口：Semaphore 提供了 acquire 和 release 方法来获取和释放资源，这样可以更直观地控制资源的访问。而"享元模式"下的实现需要手动管理线程的等待和唤醒，使用 wait 和 notify 的机制更为复杂，可读性较差。
并发控制：Semaphore 可以灵活地控制并发线程的数量，通过控制许可证的数量来限制同时访问资源的线程数量。而"享元模式"中的 wait 和 notify 机制需要手动管理线程的等待和唤醒，容易出现死锁和同步问题。
性能优化：Semaphore 可以通过设置初始许可证数量、公平性等参数来进行性能优化，而"享元模式"中的 wait 和 notify 更多地依赖于程序员手动编写的同步逻辑，容易出现性能瓶颈和难以调试的问题。
可维护性：Semaphore 提供了一个一致的、标准的接口，易于理解和维护。而"享元模式"下的实现需要程序员手动管理线程的等待和唤醒，代码复杂度高，可维护性差。
因此，Semaphore 实现简单连接池在性能和可读性上更优，它提供了更直观、简洁和安全的方式来管理并发访问资源。同时，Semaphore 对并发的控制更为灵活，使得整个连接池的管理更加高效和可靠。
class Pool {
// 1. 连接池大小
private final int poolSize;
// 2. 连接对象数组
private Connection[] connections;
// 3. 连接状态数组 0 表示空闲， 1 表示繁忙
private AtomicIntegerArray states;
private Semaphore semaphore;
// 4. 构造方法初始化
public Pool(int poolSize) {
this.poolSize = poolSize;
// 让许可数与资源数一致
this.semaphore = new Semaphore(poolSize);
this.connections = new Connection[poolSize];
this.states = new AtomicIntegerArray(new int[poolSize]);
for (int i = 0; i < poolSize; i++) {
connections[i] = new MockConnection("连接" + (i+1));
}
}
// 5. 借连接
public Connection borrow() {// t1, t2, t3
// 获取许可
try {
semaphore.acquire(); // 没有许可的线程，在此等待
} catch (InterruptedException e) {
e.printStackTrace();
}
for (int i = 0; i < poolSize; i++) {
// 获取空闲连接
if(states.get(i) == 0) {
if (states.compareAndSet(i, 0, 1)) {
log.debug("borrow {}", connections[i]);
return connections[i];
}
}
}
// 不会执行到这里
return null;
}
// 6. 归还连接
public void free(Connection conn) {
for (int i = 0; i < poolSize; i++) {
if (connections[i] == conn) {
states.set(i, 0);
log.debug("free {}", conn);
semaphore.release();
break;
}
}
}
}
Semaphore 原理
加锁解锁流程
Semaphore 有点像一个停车场，permits 就好像停车位数量，当线程获得了 permits 就像是获得了停车位，然后
停车场显示空余车位减一
刚开始，permits（state）为 3，这时 5 个线程来获取资源
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898L;
NonfairSync(int permits) {
super(permits);
}
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
}
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 1192457210091910933L;
// 核心在这里将state设置进来
Sync(int permits) {
setState(permits);
}
final int getPermits() {
return getState();
}
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
假设其中 Thread-1，Thread-2，Thread-4 cas 竞争成功，而 Thread-0 和 Thread-3 竞争失败，进入 AQS 队列
park 阻塞
以Thread1为例：
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 然后又调用 tryAcquireShared
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
// 以我们当前的例子来说，其实是符合的，3-1 = 2，返回2 不小于0 所以加锁成功
// 同理 thread 1 2 3都是一样的
return remaining;
}
}
// 如果竞争失败了呢？
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 和前面一样的流程，先设置头结点
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
// 尝试再获取一次
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 然后走到这里来 park
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
这时 Thread-4 释放了 permits，状态如下
public void release() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
// 释放了之后进入 doReleaseShared方法
doReleaseShared();
return true;
}
return false;
}
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
// unpark唤醒
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
// 唤醒了以后
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 和前面一样的流程，先设置头结点
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
// 尝试再获取一次
// 唤醒了之后，再次for循环就能到这里
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 然后走到这里来 park
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
// 唤醒了以后重新设置头结点即可
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}
接下来 Thread-0 竞争成功，permits 再次设置为 0，设置自己为 head 节点，断开原来的 head 节点，unpark 接
下来的 Thread-3 节点，但由于 permits 是 0，因此 Thread-3 在尝试不成功后再次进入 park 状态
CountdownLatch
用来进行线程同步协作，等待所有线程完成倒计时。
其中构造参数用来初始化等待计数值，await() 用来等待计数归零(等待归零，只有归零了才能继续运行)，countDown() 用来让计数减一
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(3);
new Thread(() -> {
log.debug("begin...");
sleep(1);
latch.countDown();
log.debug("end...{}", latch.getCount());
}).start();
new Thread(() -> {
log.debug("begin...");
sleep(2);
latch.countDown();
log.debug("end...{}", latch.getCount());
}).start();
new Thread(() -> {
log.debug("begin...");
sleep(1.5);
latch.countDown();
log.debug("end...{}", latch.getCount());
}).start();
log.debug("waiting...");
latch.await();
log.debug("wait end...");
}
输出
18:44:00.778 c.TestCountDownLatch [main] - waiting...
18:44:00.778 c.TestCountDownLatch [Thread-2] - begin...
18:44:00.778 c.TestCountDownLatch [Thread-0] - begin...
18:44:00.778 c.TestCountDownLatch [Thread-1] - begin...
18:44:01.782 c.TestCountDownLatch [Thread-0] - end...2
18:44:02.283 c.TestCountDownLatch [Thread-2] - end...1
18:44:02.782 c.TestCountDownLatch [Thread-1] - end...0
18:44:02.782 c.TestCountDownLatch [main] - wait end...
源码
public class CountDownLatch {
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
// 这里面去判断获得锁的条件 等不等于0，等于0就相当于获得了这个锁
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
// 如果其他线程调用 realease
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
// 什么时候 成为0了，就唤醒被阻塞的线程
return nextc == 0;
}
}
}
...
}
可以配合线程池使用，改进如下
在这里说明一下，其实join也可以达到这样的效果，但是join是属于比较底层的api，而countDownlatch属于比较高级的api。
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(3);
ExecutorService service = Executors.newFixedThreadPool(4);
service.submit(() -> {
log.debug("begin...");
sleep(1);
latch.countDown();
log.debug("end...{}", latch.getCount());
});
service.submit(() -> {
log.debug("begin...");
sleep(1.5);
latch.countDown();
log.debug("end...{}", latch.getCount());
});
service.submit(() -> {
log.debug("begin...");
sleep(2);
latch.countDown();
log.debug("end...{}", latch.getCount());
});
service.submit(()->{
try {
log.debug("waiting...");
latch.await();
log.debug("wait end...");
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
输出
18:52:25.831 c.TestCountDownLatch [pool-1-thread-3] - begin...
18:52:25.831 c.TestCountDownLatch [pool-1-thread-1] - begin...
18:52:25.831 c.TestCountDownLatch [pool-1-thread-2] - begin...
18:52:25.831 c.TestCountDownLatch [pool-1-thread-4] - waiting...
18:52:26.835 c.TestCountDownLatch [pool-1-thread-1] - end...2
18:52:27.335 c.TestCountDownLatch [pool-1-thread-2] - end...1
18:52:27.835 c.TestCountDownLatch [pool-1-thread-3] - end...0
18:52:27.835 c.TestCountDownLatch [pool-1-thread-4] - wait end...
应用之同步等待多线程准备完毕
以经典的王者荣耀等待为例子，准备阶段，必须要十个人都准备好了，游戏才能开始，那么就这样来模拟
AtomicInteger num = new AtomicInteger(0);
ExecutorService service = Executors.newFixedThreadPool(10, (r) -> {
return new Thread(r, "t" + num.getAndIncrement());
});
CountDownLatch latch = new CountDownLatch(10);
String[] all = new String[10];
Random r = new Random();
for (int j = 0; j < 10; j++) {
int x = j;
service.submit(() -> {
for (int i = 0; i <= 100; i++) {
try {
// 因为加的是随机输出，所以会出现差异值
Thread.sleep(r.nextInt(100));
} catch (InterruptedException e) {
}
all[x] = Thread.currentThread().getName() + "(" + (i + "%") + ")";
System.out.print("\r" + Arrays.toString(all));
}
latch.countDown();
});
}
latch.await();
System.out.println("\n游戏开始...");
service.shutdown();
中间输出
[t0(52%), t1(47%), t2(51%), t3(40%), t4(49%), t5(44%), t6(49%), t7(52%), t8(46%), t9(46%)]
最后输出
[t0(100%), t1(100%), t2(100%), t3(100%), t4(100%), t5(100%), t6(100%), t7(100%), t8(100%), t9(100%)]
游戏开始...
应用之同步等待多个远程调用结束
@RestController
public class TestCountDownlatchController {
@GetMapping("/order/{id}")
public Map<String, Object> order(@PathVariable int id) {
HashMap<String, Object> map = new HashMap<>();
map.put("id", id);
map.put("total", "2300.00");
sleep(2000);
return map;
}
@GetMapping("/product/{id}")
public Map<String, Object> product(@PathVariable int id) {
HashMap<String, Object> map = new HashMap<>();
if (id == 1) {
map.put("name", "小爱音箱");
map.put("price", 300);
} else if (id == 2) {
map.put("name", "小米手机");
map.put("price", 2000);
}
map.put("id", id);
sleep(1000);
return map;
}
@GetMapping("/logistics/{id}")
public Map<String, Object> logistics(@PathVariable int id) {
HashMap<String, Object> map = new HashMap<>();
map.put("id", id);
map.put("name", "中通快递");
sleep(2500);
return map;
}
private void sleep(int millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
rest 远程调用 CountDownLatch
RestTemplate restTemplate = new RestTemplate();
log.debug("begin");
ExecutorService service = Executors.newCachedThreadPool();
CountDownLatch latch = new CountDownLatch(4);
service.submit(() -> {
Map<String, Object> r =
restTemplate.getForObject("http://localhost:8080/order/{1}", Map.class, 1);
latch.CountDown();
log.debug(r);
});
service.submit(() -> {
Map<String, Object> r =
restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 1);
latch.CountDown();
log.debug(r);
});
= service.submit(() -> {
Map<String, Object> r =
restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 2);
latch.CountDown();
log.debug(r);
});
= service.submit(() -> {
Map<String, Object> r =
restTemplate.getForObject("http://localhost:8080/logistics/{1}", Map.class, 1);
latch.CountDown();
log.debug(r);
});
latch.await();
log.debug("执行完毕");
service.shutdown();
rest 远程调用 future 带返回值
RestTemplate restTemplate = new RestTemplate();
log.debug("begin");
ExecutorService service = Executors.newCachedThreadPool();
CountDownLatch latch = new CountDownLatch(4);
Future<Map<String,Object>> f1 = service.submit(() -> {
Map<String, Object> r =
restTemplate.getForObject("http://localhost:8080/order/{1}", Map.class, 1);
return r;
});
Future<Map<String, Object>> f2 = service.submit(() -> {
Map<String, Object> r =
restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 1);
return r;
});
Future<Map<String, Object>> f3 = service.submit(() -> {
Map<String, Object> r =
restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 2);
return r;
});
Future<Map<String, Object>> f4 = service.submit(() -> {
Map<String, Object> r =
restTemplate.getForObject("http://localhost:8080/logistics/{1}", Map.class, 1);
return r;
});
System.out.println(f1.get());
System.out.println(f2.get());
System.out.println(f3.get());
System.out.println(f4.get());
log.debug("执行完毕");
service.shutdown();
执行结果会比同步快非常多
但是需要注意的是，没有返回结果的时候，用countdownlatch比较合适，如果有返回值结果的话，还有用future
CyclicBarrier
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(3);
ExecutorService service = Executors.newFixedThreadPool(5);
service.submit(() -> {
log.debug("task1 start...");
sleep(1);
latch.countDown();
});
service.submit(() -> {
log.debug("task2 start...");
sleep(2);
latch.countDown();
});
try{
latch.await();
}catch(InterruptedException e){
e.printStackTrace();
}
log.debug("task1 task2 finish ...");
service.shutdown();
}
但是目前的需求是 task1 task2 被反复的运行三遍
最简单的办法就是将上述的代码放置到for循环中去
public static void main(String[] args) throws InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(5);
for(int i = 0 ; i < 3 ; i++){
CountDownLatch latch = new CountDownLatch(3);
service.submit(() -> {
log.debug("task1 start...");
sleep(1);
latch.countDown();
});
service.submit(() -> {
log.debug("task2 start...");
sleep(2);
latch.countDown();
});
try{
latch.await();
}catch(InterruptedException e){
e.printStackTrace();
}
}
log.debug("task1 task2 finish ...");
service.shutdown();
}
这样依然能够实现功能，但是，其实CountDownLatch也被创建了三次。
那能不能重用呢？这个CountDownLatch只能在构造方法的时候，给一个初始值，以后就不能改了。
[ˈsaɪklɪk ˈbæriɚ] 循环栅栏，用来进行线程协作，等待线程满足某个计数。构造时设置『计数个数』，每个线程执
行到某个需要“同步”的时刻调用 await() 方法进行等待，当等待的线程数满足『计数个数』时，继续执行
CyclicBarrier cb = new CyclicBarrier(2); // 个数为2时才会继续执行
ExecutorService service = Executors.newFixedThreadPool(2);
service.submit(()->{
System.out.println("线程1开始.."+new Date());
try {
cb.await(); // 当个数不足时，等待
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("线程1继续向下运行..."+new Date());
});
service.submit(()->{
System.out.println("线程2开始.."+new Date());
try { Thread.sleep(2000); } catch (InterruptedException e) { }
try {
cb.await(); // 2 秒后，线程个数够2，继续运行
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("线程2继续向下运行..."+new Date());
});
当await等于0的时候，此时就能够继续往下运行了。
CyclicBarrier cb = new CyclicBarrier(2，()->{
// 等到两个都执行完成后，就会执行这里面的方法
log.debug("finish");
}); // 个数为2时才会继续执行
ExecutorService service = Executors.newFixedThreadPool(2);
service.submit(()->{
System.out.println("线程1开始.."+new Date());
try {
cb.await(); // 当个数不足时，等待
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("线程1继续向下运行..."+new Date());
});
service.submit(()->{
System.out.println("线程2开始.."+new Date());
try { Thread.sleep(2000); } catch (InterruptedException e) { }
try {
cb.await(); // 2 秒后，线程个数够2，继续运行
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("线程2继续向下运行..."+new Date());
});
再次调用await就会恢复成2，在一个for循环中调用
CyclicBarrier cb = new CyclicBarrier(2，()->{
// 等到两个都执行完成后，就会执行这里面的方法
log.debug("finish");
}); // 个数为2时才会继续执行
ExecutorService service = Executors.newFixedThreadPool(2);
for(int i = 0; i < 3 ; i++){
service.submit(()->{
System.out.println("线程1开始.."+new Date());
try {
cb.await(); // 当个数不足时，等待
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("线程1继续向下运行..."+new Date());
});
service.submit(()->{
System.out.println("线程2开始.."+new Date());
try { Thread.sleep(2000); } catch (InterruptedException e) { }
try {
cb.await(); // 2 秒后，线程个数够2，继续运行
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("线程2继续向下运行..."+new Date());
});
}
这样就达到了CountDownLatch的效果啦
在这里面需要注意一下，就是 希望线程池的线程数 和 屏障数是一致的
如果不一致，比如说线程池是 三个核心线程，那么当继续走for循环的时候，由于cyclicbarrier的特性，此时就是两个task1 执行完成了，而task2比较慢。这样有可能会影响最终的效果。
![Linux shell编程学习笔记23：[] [[]]的用法小结](https://img-blog.csdnimg.cn/b5369f7217264cd3bdd596e13a885b38.png)
