Java并发之semaphore(信号量)

文章目录

- 1:官方解读
- 2:通俗易懂的例子解析
- 3:代码解析
- 4:Semaphore的应用
- 5:类结构和相关方法
- - (1):类结构
  - (2):acquire()方法
  - (3):release()方法
- 6:总结

1:官方解读

semaphore信号量就是并发工具类,Semaphore管理着一组许可permit，许可的初始数量通过构造函数设定。
当线程要访问共享资源时，需要先通过acquire()方法获取许可。获取到之后许可就被当前线程占用了，在归还许可之前其他线程不能获取这个许可。调用acquire()方法时，如果没有许可可用了，就将线程阻塞，等待有许可被归还了再执行。
当执行完业务功能后，需要通过release()方法将许可证归还，以便其他线程能够获得许可证继续执行。

2:通俗易懂的例子解析

我们假设停车场仅有3个停车位，停车位就是有限的共享资源，许可数为3。一开始停车场没有车辆所有车位全部空着，然后先后到来三辆车，停车场车位够，安排进去停车。之后来的车必须在外面候着，直到停车场有空车位。当停车场有车开出去，里面有空位了，则安排一辆车进去（至于是哪辆要看选择的机制是公平还是非公平）。

从程序角度看，停车场就相当于有限的公共资源，许可数为3，车辆就相当于线程。当来一辆车时，许可数就会减1，当停车场没有车位了（许可数为0），其他来的车辆需要在外面等候着。如果有一辆车开出停车场，许可数+1，然后放进来一辆车。

3:代码解析

 public static void main(String[] args) {Semaphore semaphore = new Semaphore(3);for (int i = 0; i < 10; i++) {new Thread(()->{try {semaphore.acquire();} catch (InterruptedException e) {e.printStackTrace();}try {System.out.println(Thread.currentThread().getName()+"开始执行");Thread.sleep(1000);System.out.println(Thread.currentThread().getName()+"执行完毕");} catch (InterruptedException e) {e.printStackTrace();} finally {semaphore.release();}}).start();}}

在这里插入图片描述

4:Semaphore的应用

Semaphore可以用于做流量控制，特别是公共资源有限的应用场景，比如数据库连接。假如有多个线程读取数据后，需要将数据保存在数据库中，而可用的最大数据库连接只有10个，这时候就需要使用Semaphore来控制能够并发访问到数据库连接资源的线程个数最多只有10个。在限制资源使用的应用场景下，Semaphore是特别合适的。

5:类结构和相关方法

(1):类结构

Semaphore同样是由AQS实现的，用内部类Sync来管理锁，Sync有两个实现，分别为NonfairSync（非公平锁）和FairSync（公平锁）。
这个类结构有没有似曾相识的感觉，重入锁ReentrantLock也是同样的类结构，Semaphore的源码跟ReentrantLock有很多相似但又比ReentrantLock简单。

(2):acquire()方法

acquire()方法就是获取许可，获取到许可就可以继续执行访问共享资源，获取不到就阻塞等待其他线程归还许可。
在这里插入图片描述

入队操作

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 GCfailed = false;return;}}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}}

(3):release()方法

release()方法归还许可，其实就是将AQS.state加1。归还成功，唤醒AQS队列中等锁的线程，从被阻塞的位置开始执行。

 public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {//释放放完毕后  doReleaseShared();// 要去唤醒新的结点线程return true;}return false;}

去释放结点

protected final boolean tryReleaseShared(int releases) {for (;;) {int current = getState();int next = current + releases;if (next < current) // overflowthrow 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;//获取结点的状态  为-1表示的是等待状态if (ws == Node.SIGNAL) {if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))continue;            // loop to recheck casesunparkSuccessor(h);}else if (ws == 0 &&!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue;                // loop on failed CAS}if (h == head)                   // loop if head changedbreak;}}