信号量Semaphore的介绍

我们以一个停车场运作为例来说明信号量的作用。假设停车场只有三个车位，一开始三个车位都是空的。这时如果同时来了三辆车，看门人允许其中它们进入进入，然后放下车拦。以后来的车必须在入口等待，直到停车场中有车辆离开。这时，如果有一辆车离开停车场，看门人得知后，打开车拦，放入一辆，如果又离开一辆，则又可以放入一辆，如此往复。

在这个停车场系统中，车位是公共资源，每辆车好比一个线程，看门人起的就是信号量的作用。信号量是一个非负整数，表示了当前公共资源的可用数目（在上面的例子中可以用空闲的停车位类比信号量），当一个线程要使用公共资源时（在上面的例子中可以用车辆类比线程），首先要查看信号量，如果信号量的值大于1，则将其减1，然后去占有公共资源。如果信号量的值为0，则线程会将自己阻塞，直到有其它线程释放公共资源。

在信号量上我们定义两种操作： acquire（获取） 和 release（释放）。当一个线程调用acquire操作时，它要么通过成功获取信号量（信号量减1），要么一直等下去，直到有线程释放信号量，或超时。release（释放）实际上会将信号量的值加1，然后唤醒等待的线程。 信号量主要用于两个目的，一个是用于多个共享资源的互斥使用，另一个用于并发线程数的控制。

示例

Semaphore包含公平锁与非公平锁2种实现

非公平锁

package main.java.study; import java.util.Calendar; import java.util.Date; import java.util.concurrent.*; public class SemaphoreDemo { final static Semaphore sema = new Semaphore(5); public static class Runner implements Runnable { int i; public Runner(int i) { this.i = i; } @Override public void run() { try { sema.acquire(); Thread.sleep(1000); System.out.println( "Runner【" + i + "】:" + new Date().toString() + " done!"); } catch ( InterruptedException ex) { ex.printStackTrace(); } sema.release(); } } public static void main (String[] args) { ExecutorService service = Executors.newFixedThreadPool(10); for(int i = 0 ;i < 20;i++) { service.execute(new Runner(i)); } } }

结果如下：

Runner【2】:Sat May 25 14:27:55 CST 2019 done! Runner【0】:Sat May 25 14:27:55 CST 2019 done! Runner【4】:Sat May 25 14:27:55 CST 2019 done! Runner【3】:Sat May 25 14:27:55 CST 2019 done! Runner【1】:Sat May 25 14:27:55 CST 2019 done! Runner【10】:Sat May 25 14:27:56 CST 2019 done! Runner【12】:Sat May 25 14:27:56 CST 2019 done! Runner【11】:Sat May 25 14:27:56 CST 2019 done! Runner【13】:Sat May 25 14:27:56 CST 2019 done! Runner【14】:Sat May 25 14:27:56 CST 2019 done! Runner【5】:Sat May 25 14:27:57 CST 2019 done! Runner【15】:Sat May 25 14:27:57 CST 2019 done! Runner【7】:Sat May 25 14:27:57 CST 2019 done! Runner【18】:Sat May 25 14:27:57 CST 2019 done! Runner【19】:Sat May 25 14:27:57 CST 2019 done! Runner【9】:Sat May 25 14:27:58 CST 2019 done! Runner【6】:Sat May 25 14:27:58 CST 2019 done! Runner【16】:Sat May 25 14:27:58 CST 2019 done! Runner【17】:Sat May 25 14:27:58 CST 2019 done! Runner【8】:Sat May 25 14:27:58 CST 2019 done!

 公平锁

package main.java.study; import java.util.Calendar; import java.util.Date; import java.util.concurrent.*; public class SemaphoreDemo { final static Semaphore sema = new Semaphore(5,true); public static class Runner implements Runnable { int i; public Runner(int i) { this.i = i; } @Override public void run() { try { sema.acquire(); Thread.sleep(1000); System.out.println( "Runner【" + i + "】:" + new Date().toString() + " done!"); } catch ( InterruptedException ex) { ex.printStackTrace(); } sema.release(); } } public static void main (String[] args) { ExecutorService service = Executors.newFixedThreadPool(10); for(int i = 0 ;i < 20;i++) { service.execute(new Runner(i)); } } }

结果如下：

Runner【2】:Sat May 25 14:33:49 CST 2019 done! Runner【1】:Sat May 25 14:33:49 CST 2019 done! Runner【3】:Sat May 25 14:33:49 CST 2019 done! Runner【0】:Sat May 25 14:33:49 CST 2019 done! Runner【4】:Sat May 25 14:33:49 CST 2019 done! Runner【5】:Sat May 25 14:33:51 CST 2019 done! Runner【6】:Sat May 25 14:33:51 CST 2019 done! Runner【8】:Sat May 25 14:33:51 CST 2019 done! Runner【9】:Sat May 25 14:33:51 CST 2019 done! Runner【7】:Sat May 25 14:33:51 CST 2019 done! Runner【10】:Sat May 25 14:33:53 CST 2019 done! Runner【12】:Sat May 25 14:33:53 CST 2019 done! Runner【11】:Sat May 25 14:33:53 CST 2019 done! Runner【13】:Sat May 25 14:33:53 CST 2019 done! Runner【14】:Sat May 25 14:33:53 CST 2019 done! Runner【15】:Sat May 25 14:33:55 CST 2019 done! Runner【17】:Sat May 25 14:33:55 CST 2019 done! Runner【19】:Sat May 25 14:33:55 CST 2019 done! Runner【16】:Sat May 25 14:33:55 CST 2019 done! Runner【18】:Sat May 25 14:33:55 CST 2019 done!

源码

Semaphore：

public Semaphore(int permits) { sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); } public boolean tryAcquire() { return sync.nonfairTryAcquireShared(1) >= 0; } public boolean tryAcquire(int permits) { if (permits < 0) throw new IllegalArgumentException(); return sync.nonfairTryAcquireShared(permits) >= 0; } public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public void acquireUninterruptibly() { sync.acquireShared(1); } 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); } //释放1个许可 public void release() { sync.releaseShared(1); } //释放指定数量许可 public void release(int permits) { if (permits < 0) throw new IllegalArgumentException(); sync.releaseShared(permits); } //减少许可 protected void reducePermits(int reduction) { if (reduction < 0) throw new IllegalArgumentException(); sync.reducePermits(reduction); } //清空许可 public int drainPermits() { return sync.drainPermits(); }

Semaphore可以：

acqure 1个或者多个许可release 1个或者多个许可默认是可中断acquire默认是非公平锁

Sync：

abstract static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 1192457210091910933L; Sync(int permits) { setState(permits); } final int getPermits() { return getState(); } final int nonfairTryAcquireShared(int acquires) { for (;;) { int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) 【1】 return remaining; } } protected final boolean tryReleaseShared(int releases) { for (;;) { int current = getState(); int next = current + releases; if (next < current) // overflow 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; } } }

Fair和NoFair

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); } } /** * Fair version */ 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()) //【2】 return -1; int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } } } public final boolean hasQueuedPredecessors() { // The correctness of this depends on head being initialized // before tail and on head.next being accurate if the current // thread is first in queue. Node t = tail; // Read fields in reverse initialization order Node h = head; Node s; //队列为空，或者head节点后置为null，表示本线程未在等待队列 return h != t && ( (s = h.next) == null //head节点的后置节点为null || s.thread != Thread.currentThread() //或者后置节点关联线程不为本线程 ); }

 

比较：

非公平锁，只要许可满足就acquire成功（见【1】）公平锁，仅当本线程在队列中没有前置节点时，才acquire成功（见【2】）

 

AbstractQueuedSynchronizer源码参考：https://blog.csdn.net/demon7552003/article/details/90105335