JAVA-Lock解析-七-Semaphore解析

信号量 Semaphore 是一个并发工具类，用来控制可同时并发的线程数，其内部维护了一组虚拟许可，通过构造器指定许可的数量，每次线程执行操作时先通过acquire方法获得许可，执行完毕再通过release方法释放许可。如果无可用许可，那么 acquire 方法将一直阻塞，直到其它线程释放许可。

Semaphore 通常用于线程对统一资源的访问控制。需要注意的是它没有实现 Lock 接口，也不支持 Condition。官方示例：

 class Pool {
   private static final int MAX_AVAILABLE = 100;
   private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);

   public Object getItem() throws InterruptedException {
     available.acquire();
     return getNextAvailableItem();
   }

   public void putItem(Object x) {
     if (markAsUnused(x))
       available.release();
   }

   // Not a particularly efficient data structure; just for demo

   protected Object[] items = ... whatever kinds of items being managed
   protected boolean[] used = new boolean[MAX_AVAILABLE];

   protected synchronized Object getNextAvailableItem() {
     for (int i = 0; i < MAX_AVAILABLE; ++i) {
       if (!used[i]) {
          used[i] = true;
          return items[i];
       }
     }
     return null; // not reached
   }

   protected synchronized boolean markAsUnused(Object item) {
     for (int i = 0; i < MAX_AVAILABLE; ++i) {
       if (item == items[i]) {
          if (used[i]) {
            used[i] = false;
            return true;
          } else
            return false;
       }
     }
     return false;
   }
 }

在获取项目之前，每个线程必须从信号灯获取许可，以确保可以使用该项目。线程完成该项目后，将其返回到池中，并向信号量返回一个许可，从而允许另一个线程获取该项目。

Semaphore 内部使用了共享锁，没有独占锁和 ReentrantLock 正好相反，还有一个 ReadLock in ReentrantReadWriteLock 的区别是 Semaphore 可以在构造方法中指定共享锁的数量，ReadLock 只能是默认的 65535，而且大于 65535 还会抛出异常。Semaphore 在大于指定的共享锁数量时会阻塞。

同样 Semaphore 有三个静态内部类 Sync，NonFairSync，FairSync。

abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;

        Sync(int permits) {
            setState(permits);
        }

        final int getPermits() {
            return getState();
        }
        
        //调用过程：
        //Semaphore#acquire-> NonFairSync#tryAcquireShared -> 本方法
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {//采用自旋直到 没有空闲的共享锁或 CAS 修改 state 的值成功
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
        
        //调用过程：
        //Semaphore#release -> AQS#releaseShared -> 本方法
        protected final boolean tryReleaseShared(int releases) {
            for (;;) {//自旋直到 CAS 成功
                int current = getState();
                int next = current + releases;
                if (next < current) // overflow，表示 releases 不能是一个负值
                    throw new Error("Maximum permit count exceeded");
                if (compareAndSetState(current, next))
                    return true;
            }
        }
        
        //减少许可证的数量
        final void reducePermits(int reductions) {
            for (;;) {
                int current = getState();
                int next = current - reductions;
                if (next > current) // underflow
                    throw new Error("Permit count underflow");
                if (compareAndSetState(current, next))
                    return;
            }
        }
        
        //把当前的可用的许可证清空
        final int drainPermits() {
            for (;;) {
                int current = getState();
                if (current == 0 || compareAndSetState(current, 0))
                    return current;
            }
        }
    }

    /**
     * 非公平锁实现
     */
    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);
        }
    }

    /**
     * 公平锁实现
     */
    static final class FairSync extends Sync {
        private static final long serialVersionUID = 2014338818796000944L;

        FairSync(int permits) {
            super(permits);
        }
        
        protected int tryAcquireShared(int acquires) {
            for (;;) {
                if (hasQueuedPredecessors())
                    return -1;
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }