Java并发编程Semaphore计数信号量详解

 package me.socketthread; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; public class SemaphoreLearn { //信号量总数 private static final int SEM_MAX = 12; public static void main(String[] args) { Semaphore sem = new Semaphore(SEM_MAX); //创建线程池 ExecutorService threadPool = Executors.newFixedThreadPool(3); //在线程池中执行任务 threadPool.execute(new MyThread(sem, 7)); threadPool.execute(new MyThread(sem, 4)); threadPool.execute(new MyThread(sem, 2)); //关闭池 threadPool.shutdown(); } } class MyThread extends Thread { private volatile Semaphore sem;  // 信号量 private int count;    // 申请信号量的大小 MyThread(Semaphore sem, int count) { this.sem = sem; this.count = count; } public void run() { try { // 从信号量中获取count个许可 sem.acquire(count); Thread.sleep(2000); System.out.println(Thread.currentThread().getName() + " acquire count="+count); } catch (InterruptedException e) { e.printStackTrace(); } finally { // 释放给定数目的许可，将其返回到信号量。 sem.release(count); System.out.println(Thread.currentThread().getName() + " release " + count + ""); } } } 

 pool-1-thread-2 acquire count=4 pool-1-thread-1 acquire count=7 pool-1-thread-1 release 7 pool-1-thread-2 release 4 pool-1-thread-3 acquire count=2 pool-1-thread-3 release 2 

 Semaphore sem = new Semaphore(12);//简单来说就是给锁标识位state赋值为12 

 Semaphore.acquire(n);//从锁标识位state中获取n个信号量，简单来说是state = state-n 此时state大于0表示可以获取信号量，如果小于0则将线程阻塞 

 public void acquire(int permits) throws InterruptedException { if (permits <0) throw new IllegalArgumentException(); //获取锁 sync.acquireSharedInterruptibly(permits); } 

 public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (tryAcquireShared(arg) <0)//tryAcquireShared中尝试获取锁资源 doAcquireSharedInterruptibly(arg); //将线程阻塞 } 

 protected int tryAcquireShared(int acquires) { for (;;) { if (hasQueuedPredecessors()) return -1; //获取state值 int available = getState(); //从state中获取信号量 int remaining = available - acquires; if (remaining <0 || compareandsetstate(available, remaining))>

 /会将线程添加到FIFO队列中，并阻塞 private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { //将线程添加到FIFO队列中 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 GC failed = false; return; } } //parkAndCheckInterrupt完成线程的阻塞操作 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } 

 public void release(int permits) { if (permits <0) throw new IllegalArgumentException(); //state = state+permits，并将FIFO队列中的阻塞线程唤起 sync.releaseShared(permits); } 

 public final boolean releaseShared(int arg) { //tryReleaseShared将state设置为state = state+arg if (tryReleaseShared(arg)) { //唤起FIFO队列中的阻塞线程 doReleaseShared(); return true; } return false; } 

 protected final boolean tryReleaseShared(int releases) { for (;;) { int current = getState(); int next = current + releases; if (next 

 private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue;      // loop to recheck cases //完成阻塞线程的唤起操作 unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue;        // loop on failed CAS } if (h == head)          // loop if head changed break; } }  

 public class Semaphore implements java.io.Serializable { private static final long serialVersionUID = -3222578661600680210L; private final Sync sync; abstract static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 1192457210091910933L; //设置锁标识位state的初始值 Sync(int permits) { setState(permits); } //获取锁标识位state的值，如果state值大于其需要的值则表示锁可以获取 final int getPermits() { return getState(); } //获取state值减去acquires后的值，如果大于等于0则表示锁可以获取 final int nonfairTryAcquireShared(int acquires) { for (;;) { int available = getState(); int remaining = available - acquires; if (remaining <0 || compareandsetstate(available, remaining)) return remaining; }> 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; }>= 0; } public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } public void release() { sync.releaseShared(1); } //获取permits值锁 public void acquire(int permits) throws InterruptedException { if (permits <0) throw new IllegalArgumentException(); sync.acquireSharedInterruptibly(permits); } public void acquireUninterruptibly(int permits) { if (permits <0) throw new IllegalArgumentException(); sync.acquireShared(permits); } public boolean tryAcquire(int permits) { if (permits <0) throw new IllegalArgumentException(); return sync.nonfairTryAcquireShared(permits) >= 0; } public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException { if (permits <0) throw new IllegalArgumentException(); return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout)); } //释放 public void release(int permits) { if (permits <0) throw new IllegalArgumentException(); sync.releaseShared(permits); } public int availablePermits() { return sync.getPermits(); } public int drainPermits() { return sync.drainPermits(); } protected void reducePermits(int reduction) { if (reduction <0) throw new IllegalArgumentException(); sync.reducePermits(reduction); } public boolean isFair() { return sync instanceof FairSync; } public final boolean hasQueuedThreads() { return sync.hasQueuedThreads(); } public final int getQueueLength() { return sync.getQueueLength(); } protected Collection getQueuedThreads() { return sync.getQueuedThreads(); } public String toString() { return super.toString() + "[Permits = " + sync.getPermits() + "]"; } }