一、概述
Semaphore是一个计数信号量。从概念上将，Semaphore包含一组许可证。如果有需要的话，每个acquire()方法都会阻塞，直到获取一个可用的许可证。每个release()方法都会释放持有许可证的线程，并且归还Semaphore一个可用的许可证。然而，实际上并没有真实的许可证对象供线程使用，Semaphore只是对可用的数量进行管理维护。
信号量机制是一种有限数量的共享模式锁。控制临界资源超出范围的一种手段。可用于流量控制，限制最大的并发访问数。
Semaphore是通过共享锁实现的。根据共享锁的获取原则，Semaphore分为"公平信号量"和"非公平信号量"。
1.1、使用场景
Semaphore经常用于限制获取某种资源的线程数量。为何会有信号量？因为任何资源都是有上限的，如数据库访问数量，IO瓶颈、网络最大传输量等
二、源码查看
Semaphore也是基于AQS实现的，在其内部也是通过一个内部类Sync实现同步器AQS，同样也是通过实现Sync实现公平锁和非公平锁。
"公平信号量"和"非公平信号量"的释放信号量的机制是一样的！不同的是它们获取信号量的机制：线程在尝试获取信号量许可时，对于公平信号量而言，如果当前线程不在CLH队列的头部，则排队等候；而对于非公平信号量而言，无论当前线程是不是在CLH队列的头部，它都会直接获取信号量。该差异具体的体现在，它们的tryAcquireShared()函数的实现不同。
2.1、使用
Semaphore semaphore=new Semaphore(2); semaphore.acquire(); semaphore.release();
构造函数
public Semaphore(int permits) { sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }
- permits 表示许可线程的数量
- fair 表示公平性，如果这个设为 true 的话，下次执行的线程会是等待最久的线程
自定义同步器构建的公平锁和非公平锁创建
public class Semaphore implements java.io.Serializable { private static final long serialVersionUID = -3222578661600680210L; /** All mechanics via AbstractQueuedSynchronizer subclass */ private final Sync sync; abstract static class Sync extends AbstractQueuedSynchronizer {} static final class NonfairSync extends Sync {} static final class FairSync extends Sync{} }
查看Sync同步器的构造方法
Sync(int permits) { setState(permits); }
可以看到同步器是在构造时必须传入permits，即信号量的许可证数
2.2、公平信号量获取-acquire
public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public void acquire(int permits) throws InterruptedException { if (permits < 0) throw new IllegalArgumentException(); sync.acquireSharedInterruptibly(permits); }
信号量中的acquire()获取函数，实际上是调用的AQS中的acquireSharedInterruptibly()。
查看acquireSharedInterruptibly内部实现
public final void acquireSharedInterruptibly(int arg) throws InterruptedException { // 如果线程是中断状态，则抛出异常。 if (Thread.interrupted()) throw new InterruptedException(); // 否则，尝试获取“共享锁”；获取成功则直接返回，获取失败，则通过doAcquireSharedInterruptibly()获取。 if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }
tryAcquireShared模板方法在自定义同步器中实现，Semaphore中”公平锁“对应的tryAcquireShared()实现如下：
protected int tryAcquireShared(int acquires) { for (;;) { // 判断“当前线程”是不是CLH队列中的第一个线程线程， // 若是的话，则返回-1。 if (hasQueuedPredecessors()) return -1; // 设置“可以获得的信号量的许可数” int available = getState(); // 设置“获得acquires个信号量许可之后，剩余的信号量许可数” int remaining = available - acquires; // 如果“剩余的信号量许可数>=0”，则设置“可以获得的信号量许可数”为remaining。 if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } }
说明：tryAcquireShared()的作用是尝试获取acquires个信号量许可数。
对于Semaphore而言，state表示的是“当前可获得的信号量许可数”。
下面看看AQS中doAcquireSharedInterruptibly()的实现：
private void doAcquireSharedInterruptibly(long arg) throws InterruptedException { // 创建”当前线程“的Node节点，且Node中记录的锁是”共享锁“类型；并将该节点添加到CLH队列末尾。 final Node node = addWaiter(Node.SHARED); boolean failed = true; try { for (;;) { // 获取上一个节点。 // 如果上一节点是CLH队列的表头，则”尝试获取共享锁“。 final Node p = node.predecessor(); if (p == head) { long r = tryAcquireShared(arg); if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } // 当前线程一直等待，直到获取到共享锁。 // 如果线程在等待过程中被中断过，则再次中断该线程(还原之前的中断状态)。 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
说明：doAcquireSharedInterruptibly()会使当前线程一直等待，直到当前线程获取到共享锁(或被中断)才返回。
(01) addWaiter(Node.SHARED)的作用是，创建”当前线程“的Node节点，且Node中记录的锁的类型是”共享锁“(Node.SHARED)；并将该节点添加到CLH队列末尾。
(02) node.predecessor()的作用是，获取上一个节点。如果上一节点是CLH队列的表头，则”尝试获取共享锁“。
(03) shouldParkAfterFailedAcquire()的作用和它的名称一样，如果在尝试获取锁失败之后，线程应该等待，则返回true；否则，返回false。
(04) 当shouldParkAfterFailedAcquire()返回ture时，则调用parkAndCheckInterrupt()，当前线程会进入等待状态，直到获取到共享锁才继续运行。
2.3、公平信号量释放-release
Semaphore中公平信号量(FairSync)的释放API如下：
public void release() { sync.releaseShared(1); } public void release(int permits) { if (permits < 0) throw new IllegalArgumentException(); sync.releaseShared(permits); }
信号量的releases()释放函数，实际上是调用的AQS中的releaseShared()。
releaseShared()在AQS中实现，源码如下：
public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }
说明：releaseShared()的目的是让当前线程释放它所持有的共享锁。
它首先会通过tryReleaseShared()去尝试释放共享锁。尝试成功，则直接返回；尝试失败，则通过doReleaseShared()去释放共享锁。
Semaphore重写了tryReleaseShared()，它的源码如下：
protected final boolean tryReleaseShared(int releases) { for (;;) { // 获取“可以获得的信号量的许可数” int current = getState(); // 获取“释放releases个信号量许可之后，剩余的信号量许可数” int next = current + releases; if (next < current) // overflow throw new Error("Maximum permit count exceeded"); // 设置“可以获得的信号量的许可数”为next。 if (compareAndSetState(current, next)) return true; } }
如果tryReleaseShared()尝试释放共享锁失败，则会调用doReleaseShared()去释放共享锁。doReleaseShared()的源码如下：
private void doReleaseShared() { for (;;) { // 获取CLH队列的头节点 Node h = head; // 如果头节点不为null，并且头节点不等于tail节点。 if (h != null && h != tail) { // 获取头节点对应的线程的状态 int ws = h.waitStatus; // 如果头节点对应的线程是SIGNAL状态，则意味着“头节点的下一个节点所对应的线程”需要被unpark唤醒。 if (ws == Node.SIGNAL) { // 设置“头节点对应的线程状态”为空状态。失败的话，则继续循环。 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // 唤醒“头节点的下一个节点所对应的线程”。 unparkSuccessor(h); } // 如果头节点对应的线程是空状态，则设置“文件点对应的线程所拥有的共享锁”为其它线程获取锁的空状态。 else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } // 如果头节点发生变化，则继续循环。否则，退出循环。 if (h == head) // loop if head changed break; } }
说明：doReleaseShared()会释放“共享锁”。它会从前往后的遍历CLH队列，依次“唤醒”然后“执行”队列中每个节点对应的线程；最终的目的是让这些线程释放它们所持有的信号量。
2.4、非公平信号量获取和释放
Semaphore中的非公平信号量是NonFairSync。在Semaphore中，“非公平信号量许可的释放(release)”与“公平信号量许可的释放(release)”是一样的。
不同的是它们获取“信号量许可”的机制不同，下面是非公平信号量获取信号量许可的代码。
非公平信号量的tryAcquireShared()实现如下：
protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires); }
nonfairTryAcquireShared()的实现如下：
final int nonfairTryAcquireShared(int acquires) { for (;;) { // 设置“可以获得的信号量的许可数” int available = getState(); // 设置“获得acquires个信号量许可之后，剩余的信号量许可数” int remaining = available - acquires; // 如果“剩余的信号量许可数>=0”，则设置“可以获得的信号量许可数”为remaining。 if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } }
说明：非公平信号量的tryAcquireShared()调用AQS中的nonfairTryAcquireShared()。而在nonfairTryAcquireShared()的for循环中，它都会直接判断“当前剩余的信号量许可数”是否足够；足够的话，则直接“设置可以获得的信号量许可数”，进而再获取信号量。
而公平信号量的tryAcquireShared()中，在获取信号量之前会通过if (hasQueuedPredecessors())来判断“当前线程是不是在CLH队列的头部”，是的话，则返回-1。
三、示例
public class SemaphoreTest { // 自定义工作线程 private static class Worker extends Thread { private CountDownLatch countDownLatch; private Semaphore semaphore; public Worker(CountDownLatch countDownLatch, Semaphore semaphore) { this.countDownLatch = countDownLatch; this.semaphore = semaphore; } @Override public void run() { super.run(); boolean hasAcquire = false; try { countDownLatch.await(); semaphore.acquire(); hasAcquire = true; System.out.println(Thread.currentThread().getName() + "开始执行"); // 工作线程开始处理，这里用Thread.sleep()来模拟业务处理 Thread.sleep(1000); System.out.println(Thread.currentThread().getName() + "执行完毕"); } catch (Exception e) { e.printStackTrace(); } finally { if (hasAcquire) { semaphore.release(); } } } } public static void main(String[] args) { int threadCount = 10; int permitCount = 3; CountDownLatch countDownLatch = new CountDownLatch(1); Semaphore semaphore = new Semaphore(permitCount); for (int i = 0; i < threadCount; i++) { Worker worker = new Worker(countDownLatch, semaphore); worker.start(); } countDownLatch.countDown(); } }
输出
Thread-1开始执行 Thread-0开始执行 Thread-9开始执行 Thread-1执行完毕 Thread-2开始执行 Thread-9执行完毕 Thread-0执行完毕 Thread-3开始执行 Thread-4开始执行 Thread-2执行完毕 Thread-3执行完毕 Thread-4执行完毕 Thread-6开始执行 Thread-5开始执行 Thread-7开始执行 Thread-6执行完毕 Thread-7执行完毕 Thread-5执行完毕 Thread-8开始执行 Thread-8执行完毕
Semaphore有3个许可，但是有10个线程要执行，从执行结果中可以看出，每次都是3个线程一组开始执行，也就是每次只能有3个线程获取许可。
参看地址：
http://www.cnblogs.com/skywang12345/p/3534050.html