Semaphore 信号量，可用于控制一定时间内，并发执行的线程数，基于 AQS 实现。可应用于网关限流、资源限制 (如 最大可发起连接数)。由于 release() 释放许可时，未对释放许可数做限制，所以可以通过该方法增加总的许可数量。

获取许可 支持公平和非公平模式，默认非公平模式。公平模式无论是否有许可，都会先判断是否有线程在排队，如果有线程排队，则进入排队，否则尝试获取许可；非公平模式无论许可是否充足，直接尝试获取许可。

不多废话，下面直接看源码。

核心内部类 Sync

  1. abstract static class Sync extends AbstractQueuedSynchronizer {
  3.     private static final long serialVersionUID = 1192457210091910933L;
  5.     /* 赋值state为总许可数 */
  6.     Sync(int permits) {
  7.         setState(permits);
  8.     }
  10.     /* 剩余许可数 */
  11.     final int getPermits() {
  12.         return getState();
  13.     }
  15.     /* 自旋 + CAS非公平获取 */
  16.     final int nonfairTryAcquireShared(int acquires) {
  17.         for (;;) {
  18.             // 剩余可用许可数
  19.             int available = getState();
  20.             // 本次获取许可后，剩余许可
  21.             int remaining = available - acquires;
  22.             // 如果获取后，剩余许可大于0，则CAS更新剩余许可，否则获取失败失败
  23.             if (remaining < 0 ||
  24.                 compareAndSetState(available, remaining))
  25.                 return remaining;
  26.         }
  27.     }
  29.     /**
  30.      * 自旋 + CAS 释放许可
  31.      * 由于未对释放许可数做限制，所以可以通过release动态增加许可数量
  32.      */
  33.     protected final boolean tryReleaseShared(int releases) {
  34.         for (;;) {
  35.             // 当前剩余许可
  36.             int current = getState();
  37.             // 许可更新值
  38.             int next = current + releases;
  39.             // 如果许可更新值为负数，说明许可数量溢出，抛出错误
  40.             if (next < current) // overflow
  41.                 throw new Error("Maximum permit count exceeded");
  42.             // CAS更新许可数量
  43.             if (compareAndSetState(current, next))
  44.                 return true;
  45.         }
  46.     }
  48.     /* 自旋 + CAS 减少许可数量 */
  49.     final void reducePermits(int reductions) {
  50.         for (;;) {
  51.             // 当前剩余许可
  52.             int current = getState();
  53.             // 更新值
  54.             int next = current - reductions;
  55.                // 较少许可数错误，抛出异常
  56.             if (next > current) // underflow
  57.                 throw new Error("Permit count underflow");
  58.             // CAS更新许可数
  59.             if (compareAndSetState(current, next))
  60.                 return;
  61.         }
  62.     }
  64.     /* 丢弃所有许可 */
  65.     final int drainPermits() {
  66.         for (;;) {
  67.             int current = getState();
  68.             if (current == 0 || compareAndSetState(current, 0))
  69.                 return current;
  70.         }
  71.     }
  72. }
  74. /**
  75.  * 非公平模式
  76.  */
  77. static final class NonfairSync extends Sync {
  78.     private static final long serialVersionUID = -2694183684443567898L;
  80.     NonfairSync(int permits) {
  81.         super(permits);
  82.     }
  84.     protected int tryAcquireShared(int acquires) {
  85.         return nonfairTryAcquireShared(acquires);
  86.     }
  87. }
  89. /**
  90.  * 公平模式
  91.  */
  92. static final class FairSync extends Sync {
  93.     private static final long serialVersionUID = 2014338818796000944L;
  95.     FairSync(int permits) {
  96.         super(permits);
  97.     }
  99.     /**
  100.      * 公平模式获取许可
  101.      * 公平模式不论许可是否充足，都会判断同步队列中是否有线程在等地，如果有，获取失败，排队阻塞
  102.      */
  103.     protected int tryAcquireShared(int acquires) {
  104.         for (;;) {
  105.             // 如果有线程在排队，立即返回
  106.             if (hasQueuedPredecessors())
  107.                 return -1;
  108.             // 自旋 + cas获取许可
  109.             int available = getState();
  110.             int remaining = available - acquires;
  111.             if (remaining < 0 ||
  112.                 compareAndSetState(available, remaining))
  113.                 return remaining;
  114.         }
  115.     }
  116. }

主要 API

  1. public class Semaphore implements java.io.Serializable {
  3.     private static final long serialVersionUID = -3222578661600680210L;
  5.     /** All mechanics via AbstractQueuedSynchronizer subclass */
  6.     private final Sync sync;
  8.     /**
  9.      * 根据给定的 总许可数permits，创建 Semaphore
  10.      */
  11.     public Semaphore(int permits) {
  12.         sync = new NonfairSync(permits);
  13.     }
  15.     /**
  16.      * fair为true表示使用公平锁模式，false使用非公平锁
  17.      */
  18.     public Semaphore(int permits, boolean fair) {
  19.         sync = fair ? new FairSync(permits) : new NonfairSync(permits);
  20.     }
  22.     // --------------------- 获取许可 --------------------
  24.     /* 获取指定数量的许可    */
  25.     public void acquire(int permits) throws InterruptedException {
  26.         if (permits < 0) throw new IllegalArgumentException();
  27.         sync.acquireSharedInterruptibly(permits);
  28.     }
  30.     /* 获取一个许可    */
  31.     public void acquire() throws InterruptedException {
  32.         sync.acquireSharedInterruptibly(1);
  33.     }
  35.     public final void acquireSharedInterruptibly(int arg)
  36.         throws InterruptedException {
  37.         if (Thread.interrupted())
  38.             throw new InterruptedException();
  39.         if (tryAcquireShared(arg) < 0) // 获取许可，剩余许可>=0，则获取许可成功，<0获取许可失败，进入排队
  40.             doAcquireSharedInterruptibly(arg);
  41.     }
  43.     protected int tryAcquireShared(int acquires) {
  44.         return nonfairTryAcquireShared(acquires);
  45.     }
  47.     /**
  48.      * @return 剩余许可数量。非负数，获取许可成功，负数，获取许可失败
  49.      */
  50.     final int nonfairTryAcquireShared(int acquires) {
  51.         for (;;) {
  52.             int available = getState();
  53.             int remaining = available - acquires;
  54.             if (remaining < 0 ||
  55.                 compareAndSetState(available, remaining))
  56.                 return remaining;
  57.         }
  58.     }
  60.     /**
  61.      * 获取许可失败，当前线程进入同步队列，排队阻塞
  62.      */
  63.     private void doAcquireSharedInterruptibly(int arg)
  64.         throws InterruptedException {
  65.         // 创建同步队列节点，并入队
  66.         final Node node = addWaiter(Node.SHARED);
  67.         boolean failed = true;
  68.         try {
  69.             for (;;) {
  70.                 // 如果当前节点是第二个节点，尝试获取锁
  71.                 final Node p = node.predecessor();
  72.                 if (p == head) {
  73.                     int r = tryAcquireShared(arg);
  74.                     if (r >= 0) {
  75.                         setHeadAndPropagate(node, r);
  76.                         p.next = null; // help GC
  77.                         failed = false;
  78.                         return;
  79.                     }
  80.                 }
  81.                 // 阻塞当前线程
  82.                 if (shouldParkAfterFailedAcquire(p, node) &&
  83.                     parkAndCheckInterrupt())
  84.                     throw new InterruptedException();
  85.             }
  86.         } finally {
  87.             if (failed)
  88.                 cancelAcquire(node);
  89.         }
  90.     }
  92.     // --------------------- 释放归还许可 -------------------------
  94.     /* 释放指定数量的许可 */
  95.     public void release(int permits) {
  96.         if (permits < 0) throw new IllegalArgumentException();
  97.         sync.releaseShared(permits);
  98.     }
  100.     /* 释放一个许可 */
  101.     public void release() {
  102.         sync.releaseShared(1);
  103.     }
  105.     public final boolean releaseShared(int arg) {
  106.         // 归还许可成功
  107.         if (tryReleaseShared(arg)) {
  108.             doReleaseShared();
  109.             return true;
  110.         }
  111.         return false;
  112.     }
  114.     /**
  115.      * 释放许可
  116.      * 由于未对释放许可数做限制，所以可以通过release动态增加许可数量
  117.      */
  118.     protected final boolean tryReleaseShared(int releases) {
  119.         for (;;) {
  120.             int current = getState();
  121.             int next = current + releases;
  122.             if (next < current) // overflow
  123.                 throw new Error("Maximum permit count exceeded");
  124.             if (compareAndSetState(current, next))
  125.                 return true;
  126.         }
  127.     }
  129.     private void doReleaseShared() {
  130.         // 自旋，唤醒等待的第一个线程(其他线程将由第一个线程向后传递唤醒)
  131.         for (;;) {
  132.             Node h = head;
  133.             if (h != null && h != tail) {
  134.                 int ws = h.waitStatus;
  135.                 if (ws == Node.SIGNAL) {
  136.                     if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
  137.                         continue;            // loop to recheck cases
  138.                     // 唤醒第一个等待线程
  139.                     unparkSuccessor(h);
  140.                 }
  141.                 else if (ws == 0 &&
  142.                          !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
  143.                     continue;                // loop on failed CAS
  144.             }
  145.             if (h == head)                   // loop if head changed
  146.                 break;
  147.         }
  148.     }
  149. }