使用
概述
Semaphore（信号量）是Java中一个并发控制工具，用于控制对共享资源的访问。它基于计数器的原理，可以限制同时访问某个资源的线程数量。
在Java中使用Semaphore，你需要按照以下步骤进行操作：
导包：
import java.util.concurrent.Semaphore;
创建Semaphore对象：
Semaphore semaphore = new Semaphore(n);
其中，n是允许同时访问共享资源的线程数量。
在需要访问共享资源的代码段前后，使用acquire()和release()方法来获取和释放信号量：
1. try { 2. semaphore.acquire(); // 获取信号量，如果没有可用的许可证，线程将被阻塞 3. // 访问共享资源的代码 4. } catch (InterruptedException e) { 5. // 处理中断异常 6. } finally { 7. semaphore.release(); // 释放信号量，增加一个许可证 8. }
acquire()方法尝试获取一个许可证，如果当前没有可用的许可证，则该线程将被阻塞，直到有可用的许可证为止。release()方法释放一个许可证，使其可供其他线程使用。
通过适当地使用acquire()和release()方法，在超过信号量允许的线程数量时，可以限制并发访问共享资源的线程数量，实现线程间的同步和互斥。
需要注意的是，Semaphore还提供了一些其他方法，如availablePermits()用于获取当前可用的许可证数量，以及tryAcquire()方法在不阻塞线程的情况下尝试获取许可证等。
具体例子
1. public static void main(String[] args) { 2. // 1. 创建 semaphore 对象 3. Semaphore semaphore = new Semaphore(3); 4. // 2. 10个线程同时运行 5. for (int i = 0; i < 10; i++) { 6. new Thread(() -> { 7. // 3. 获取许可 8. 9. try { 10. semaphore.acquire(); 11. } catch (InterruptedException e) { 12. e.printStackTrace(); 13. } 14. try { 15. log.debug("running..."); 16. sleep(1); 17. log.debug("end..."); 18. } finally { 19. // 4. 释放许可 20. 21. semaphore.release(); 22. } 23. }).start(); 24. } 25. 26. }
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 有点像一个停车场，permits 就好像停车位数量，当线程获得了 permits 就像是获得了停车位，然后 停车场显示空余车位减一 刚开始，permits（state）为 3，这时 5 个线程来获取资源
假设其中 Thread-1，Thread-2，Thread-4 cas 竞争成功，而 Thread-0 和 Thread-3 竞争失败，进入 AQS 队列park 阻塞
接下来 Thread-0 竞争成功，permits 再次设置为 0，设置自己为 head 节点，断开原来的 head 节点，unpark 接 下来的 Thread-3 节点，但由于 permits 是 0，因此 Thread-3 在尝试不成功后再次进入 park 状态
+
源码
构造方法有俩个：
1. public Semaphore(int permits) { 2. sync = new NonfairSync(permits); 3. }
创建 Semaphore 具有给定数量的许可和不公平公平设置。
参数：
许可证 – 可用的许可证的初始数量。此值可能为负数，在这种情况下，必须先进行释放，然后才能授予任何收购。
1. public Semaphore(int permits, boolean fair) { 2. sync = fair ? new FairSync(permits) : new NonfairSync(permits); 3. }
创建具有给定数量的许可和给定公平性设置的 。Semaphore
参数：
许可证 – 可用的许可证的初始数量。此值可能为负数，在这种情况下，必须先进行释放，然后才能授予任何收购。
公平 – true 如果此信号量将保证在争用中授予先进先出的许可证，否则 false
1. static final class NonfairSync extends Sync { 2. private static final long serialVersionUID = -2694183684443567898L; 3. 4. NonfairSync(int permits) { 5. // permits 即 state 6. 7. super(permits); 8. } 9. 10. // Semaphore 方法, 方便阅读, 放在此处 11. 12. public void acquire() throws InterruptedException { 13. sync.acquireSharedInterruptibly(1); 14. } 15. 16. // AQS 继承过来的方法, 方便阅读, 放在此处 17. 18. public final void acquireSharedInterruptibly(int arg) 19. throws InterruptedException { 20. if (Thread.interrupted()) 21. throw new InterruptedException(); 22. if (tryAcquireShared(arg) < 0) 23. doAcquireSharedInterruptibly(arg); 24. } 25. 26. // 尝试获得共享锁 27. 28. protected int tryAcquireShared(int acquires) { 29. return nonfairTryAcquireShared(acquires); 30. } 31. 32. // Sync 继承过来的方法, 方便阅读, 放在此处 33. 34. final int nonfairTryAcquireShared(int acquires) { 35. for (;;) { 36. int available = getState(); 37. int remaining = available - acquires; 38. if ( 39. // 如果许可已经用完, 返回负数, 表示获取失败, 进入 doAcquireSharedInterruptibly 40. 41. remaining < 0 || 42. 43. // 如果 cas 重试成功, 返回正数, 表示获取成功 44. 45. compareAndSetState(available, remaining) 46. ) { 47. return remaining; 48. } 49. } 50. } 51. 52. // AQS 继承过来的方法, 方便阅读, 放在此处 53. 54. private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { 55. final Node node = addWaiter(Node.SHARED); 56. boolean failed = true; 57. try { 58. for (;;) { 59. final Node p = node.predecessor(); 60. if (p == head) { 61. // 再次尝试获取许可 62. 63. int r = tryAcquireShared(arg); 64. if (r >= 0) { 65. // 成功后本线程出队（AQS）, 所在 Node设置为 head 66. 67. // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark 68. 69. // 如果 head.waitStatus == 0 ==> Node.PROPAGATE 70. 71. // r 表示可用资源数, 为 0 则不会继续传播 72. 73. setHeadAndPropagate(node, r); 74. p.next = null; // help GC 75. 76. failed = false; 77. return; 78. } 79. } 80. // 不成功, 设置上一个节点 waitStatus = Node.SIGNAL, 下轮进入 park 阻塞 81. 82. if (shouldParkAfterFailedAcquire(p, node) && 83. 84. parkAndCheckInterrupt()) 85. throw new InterruptedException(); 86. } 87. } finally { 88. if (failed) 89. cancelAcquire(node); 90. } 91. } 92. 93. // Semaphore 方法, 方便阅读, 放在此处 94. 95. public void release() { 96. sync.releaseShared(1); 97. } 98. 99. // AQS 继承过来的方法, 方便阅读, 放在此处 100. 101. public final boolean releaseShared(int arg) { 102. if (tryReleaseShared(arg)) { 103. doReleaseShared(); 104. return true; 105. } 106. return false; 107. } 108. 109. // Sync 继承过来的方法, 方便阅读, 放在此处 110. protected final boolean tryReleaseShared(int releases) { 111. for (;;) { 112. int current = getState(); 113. int next = current + releases; 114. if (next < current) // overflow 115. throw new Error("Maximum permit count exceeded"); 116. if (compareAndSetState(current, next)) 117. return true; 118. } 119. } 120. }
