线程之Semaphore示例

一. Semaphore基本介绍： 

private final int capacity = 10;
private final Semaphore empty = new Semaphore(capacity); //仓库中空的槽的信号量
private final Semaphore full = new Semaphore(0); //仓库中被占用的槽的信号量
private final Semaphore mutex = new Semaphore(1); //互斥信号量
private int insertIndex = 0; // 仓库中当前可以放置物品的位置。
private int removeIndex = 0; // 仓库中当前可以拿走物品的位置。

信号量是多线程使用的一种方式，
计数信号量(Semaphore) 维护着一个许可集。调用acquire()获取一个许可，release()释放一个许可。
设置信号量是否采用公平模式 ？ 如以公平方式执行，则线程将会按到达的顺序(FIFO)执行；如是非公平，则后请求的有可能排在队列的头部。
Semaphore在多线程环境下被扩放使用，操作系统的信号量是个很重要的概念，在进程控制方面都有应用。
同步机制信号量(Semaphore) 使用场景？

1）Semaphore可轻松完成信号量控制,可以控制多个线程同时对某个资源的访问。
用来保证两个或多个程序不能同时进入临界区，从而不能同时使用一个共享资源，达到线程互斥的作用。
Lock和synchronized是锁的互斥，一个线程如果锁定了一资源，那么其它线程只能等待资源的释放。也就是一次只有一个线程执行，等到这个线程执行完毕或者unlock。
如，厕所有5个坑，假如有10个人要上厕所，那么同时只能有多少个人去上厕所呢？
同时只能有5个人能够占用，当5个人中的任何一个人让开后，其中等待的另外5个人中又有一个人可以占用了。另外等待的5个人中可以是随机获得优先机会，也可以是按照先来后到顺序获得机会，这取决于构造Semaphore对象时传入的参数选项。
2）单个信号量Semaphore对象可实现互斥锁的功能，且可以是由一个线程获得了“锁”，再由另一个线程释放“锁”，应用于死锁恢复的一些场合。
信号量用在多线程多任务同步的，一个线程完成了某一个动作就通过信号量告诉别的线程，别的线程再进行某些动作。
也就是说Semaphore不一定是锁定某个资源，而是流程上的概念。
比方有A,B两个线程，B线程的操作可能要等A线程执行完毕之后才执行，这个任务并不一定是锁定某一资源，还可以是进行一些计算或者数据处理之类，它们也许并不访问共享变量，只是逻辑上的先后顺序。
3）信号量也可实现线程同步：用信号量实现了生产者消费者问题：
在生产者消费者问题中，信号量full和empty用来保证某种事件顺序发生或者不发生，即对于一个物品，生产的过程始终要在消费的过程之前。
里面有三个信号量full，empty和mutex，其中full和empty用来做进程同步，mutex用来做进程互斥。
每当要生产一个物品时，我们首先检查能否获取信号量empty的一个许可，如果不可以则阻塞线程，如果可以则继续获取items数组的访问权，该访问权由互斥信号量mutex来控制，当放置完物品后，会释放信号量empty的一个许可。
同理，每当要消费一个物品时，我们首先检查能否获取信号量full的一个许可，如果不可以则阻塞线程，如果可以择继续获取items数组的访问权，当拿走物品后，会释放信号量full的一个许可。

信号量和信号区别在哪里？
如果我们单单使用信号，即使用wait和notify方法，很有可能会错过丢失某些信号通知。
比如，如果我们不对count的访问添加限制，当count为0时，调度程序知道仓库里没有物品了，便准备挂起消费者线程并启动生产者线程，
如果生产者线程的启动时间比消费者线程的挂起时间快得多，很可能会有这种情况：
当生产者线程生产了一个物品，使count由0变为1，此时会向消费者线程发送一个唤醒信号，但此时消费者线程还没有完全挂起，因此它会忽略这个信号，这样一来，生产者线程会一直生产物品，直到count为capacity时挂起，而消费者线程在完全挂起之后不会再收到唤醒信号，因此也会一直挂起，这样整个系统就全部挂起，永远不会执行下去。
当然，我们的实现方式是同一时刻只能有一个线程操作count，因此消费者线程的wait方法一定在生产者线程的notify方法之前执行，即当消费者线程完全挂起之后，生产者线程才能启动，于是不会出现错过丢失信号的问题。
而Java中Semaphore的实现也是建立在信号的基础上的，但不同的是它会保存所有的信号，不会丢弃任何信号，这样就很好地避免了前面的问题，这也是信号量和信号的最大区别所在。

信号（Signal）是一种处理异步事件的通讯方式，用于通知其他进程或者自己本身，来告知将有某种事件发生。
在Java中，信号机制通过wait()，notify()和notifyAll()来实现。其中wait()使得当前调用wait()的线程挂起，并释放已经获得的wait()所在代码块的锁；
notify()用于随即唤醒一个被wait()挂起的线程进入线程调度队列；notifyAll()用于唤醒所有被wait()挂起的线程进入线程调度队列。

二.Java并发编程 — 信号量与互斥量
信号量与互斥量的区别：

信号量用于线程同步，互斥量用户保护资源的互斥访问。
互斥量用于线程的互斥，信号线用于线程的同步。
互斥量值只能为0/1，信号量值可以为非负整数。信号量可以实现多个同类资源的多线程互斥和同步。
互斥量的加锁和解锁必须由同一线程分别对应使用，信号量可以由一个线程释放，另一个线程得到。
信号量Semaphore，信号量是在多线程环境中，线程间传递信号的一种方式。

1.Java内置的Semaphore
java.util.concurrent包中有Semaphore的实现，可以设置参数，控制同时访问的个数。
下面Demo申明了一个只有5个许可的Semaphore，而有20个线程要访问这个资源，通过acquire()和release()获取和释放访问许可。

final Semaphore semp = new Semaphore(5);
ExecutorService exec = Executors.newCachedThreadPool();
for (int index = 0; index < 20; index++) {
    final int NO = index;
    Runnable run = new Runnable() {
    public void run() {
     try {
      semp.acquire();
      System.out.println("Accessing: " + NO);
      Thread.sleep((long) (Math.random() * 10000));
      // 访问完后，释放
      semp.release();
      System.out.println("-----------------" + semp.availablePermits());}}
   };
  exec.execute(run);
}
exec.shutdown();

2. 互斥量Mutex
互斥量：提供对资源的独占访问，只能为0/1，如果某一个资源同时只能允许一个访问者对其访问，可以使用互斥量控制线程对其访问。
互斥量实现：
public class Mutex {
private boolean isLocked = false;
public synchronized void lock() {
while(this.isLocked) //使用while可以避免线程 假唤醒
  wait();
  this.isLocked= true;}
}
public synchronized void unlock() throws InterruptedException{
  this.isLocked= false;
  this.notify();}
}
在Mutex中，我们添加了一个signal用于保存信号。
将互斥量当作锁来使用：
Mutex mutex = new Mutex();
mutex.lock();
...
//临界区
mutex.unlock();
互斥量的加锁和解锁必须由同一个线程分别对应使用。

三.信号量介绍：

Semaphore（信号量） 是一个线程同步结构，用于在线程间传递信号，以避免出现信号丢失，或者像锁一样用于保护一个关键区域。自从5.0开始，jdk在java.util.concurrent包里提供了Semaphore 的官方实现，因此大家不需要自己去实现Semaphore。但是还是很有必要去熟悉如何使用Semaphore及其背后的原理：
本文涉及：
简单的Semaphore实现
使用Semaphore来发出信号
可计数的Semaphore
有上限的Semaphore
把Semaphore当锁来使用

1、简单的Semaphore实现
一个信号量的简单实现：
public class Semaphore {
  private boolean signal = false;//使用signal可以避免信号丢失
  public synchronized void take() {
　　this.signal = true;
　　this.notify();}

  public synchronized void release() throws InterruptedException{
　　while(!this.signal) //使用while避免假唤醒
　　wait();
　　this.signal = false;}
}

Take方法发出一个被存放在Semaphore内部的信号，而Release方法则等待一个信号，当其接收到信号后，标记位signal被清空，然后该方法终止。
使用这个semaphore可以避免错失某些信号通知。用take方法来代替notify，release方法来代替wait。
如果某线程在调用release等待之前调用take方法，那么调用release方法的线程仍然知道take方法已经被某个线程调用过了，因为该Semaphore内部保存了take方法发出的信号。而wait和notify方法就没有这样的功能。当用semaphore来产生信号时，take和release这两个方法名看起来有点奇怪。
这两个名字来源于后面把semaphore当做锁的例子，后面会详细介绍这个例子，在该例子中，take和release这两个名字会变得很合理。
2、使用Semaphore来产生信号
下面的例子，两个线程通过Semaphore发出的信号来通知对方
Semaphore semaphore = new Semaphore();
SendingThread sender = new SendingThread(semaphore)；
ReceivingThread receiver = new ReceivingThread(semaphore);
receiver.start();
sender.start();
public class SendingThread {
Semaphore semaphore = null;
public SendingThread(Semaphore semaphore){
　　this.semaphore = semaphore;
}
public void run(){
   while(true){
　　//do something, then signal
　　this.semaphore.take();
 }}
}
public class RecevingThread {
  Semaphore semaphore = null;
  public ReceivingThread(Semaphore semaphore){
　　this.semaphore = semaphore;
}
public void run(){
   while(true){
　　this.semaphore.release();
　　//receive signal, then do something...}
}}
3、可计数的Semaphore
上面提到的Semaphore的简单实现并没有计算通过调用take方法所产生信号的数量。可以把它改造成具有计数功能的Semaphore。下面是一个可计数的Semaphore的简单实现。
public class CountingSemaphore {
private int signals = 0;
public synchronized void take() {
　　this.signals++;
　　this.notify();  }

public synchronized void release() throws InterruptedException{
　　while(this.signals == 0) wait();
　　this.signals--;}
}
4、有上限的Semaphore
上面的CountingSemaphore并没有限制信号的数量。下面的代码将CountingSemaphore改造成一个信号数量有上限的BoundedSemaphore。
public class BoundedSemaphore {
private int signals = 0;
private int bound = 0;
public BoundedSemaphore(int upperBound){
　　this.bound = upperBound;
}
public synchronized void take() throws InterruptedException{
while(this.signals == bound) wait();
　　this.signals++;
　　this.notify();
}
public synchronized void release() throws InterruptedException{
while(this.signals == 0) wait();
　　this.signals--;
　　this.notify();}
}
在BoundedSemaphore中，当已经产生的信号数量达到了上限，take方法将阻塞新的信号产生请求，直到某个线程调用release方法后，被阻塞于take方法的线程才能传递自己的信号。
5、把Semaphore当锁来使用
当信号量的数量上限是1时，Semaphore可以被当做锁来使用。通过take和release方法来保护关键区域。请看下面的例子：
BoundedSemaphore semaphore = new BoundedSemaphore(1);
...
semaphore.take();
try{
  //critical section
} finally {
  semaphore.release();
}
在前面的例子中，Semaphore被用来在多个线程之间传递信号，这种情况下，take和release分别被不同的线程调用。
但是在锁这个例子中，take和release方法将被同一线程调用，因为只允许一个线程来获取信号（允许进入关键区域的信号），其它调用take方法获取信号的线程将被阻塞，知道第一个调用take方法的线程调用release方法来释放信号。对release方法的调用永远不会被阻塞，这是因为任何一个线程都是先调用take方法，然后再调用release。
通过有上限的Semaphore可以限制进入某代码块的线程数量。设想一下，在上面例子中，如果BoundedSemaphore 上限设为5将会发生什么？意味着允许5个线程同时访问关键区域，但是你必须保证，这个5个线程不会互相冲突。否则你的应用程序将不能正常运行。
必须注意，release方法应当在finally块中被执行。这样可以保在关键区域的代码抛出异常的情况下，信号也一定会被释放。

二. 示例

为了提高接口的响应速度,可以使用ThreadPoolExecutor + Runnable 或者ThreadPoolExecutor 并发调用 技术来并行执行task。但是ThreadPoolExecutor有个特点,就是当core线程不足以应付请求的时候，会将task加入到队列中。一旦使用队列，那么就可能出现队列爆掉或者队列导致的内存溢出问题。

为了尽快提供接口响应速度，但是又不想使用队列特性的话。可以使用信号量来做到。
Semaphore信号量管理着一组许可,在执行操作时需要首先获得许可,并在使用后释放许可。如果已经没有许可了, acquire方法将一直阻塞,直到有许可。

信号量简单例子：
ThreadPoolExecutor中使用信号量：
在ThreadPoolExecutor中，我们在定义core线程参数的时候，比如定义为10个，那么使用信号量的时候，初始化参数也设置为10.
Semaphore<Integer> sem= new Semaphore<>(10);
ThreadPoolExecutor中,如果不想用到队列，就必须保证线程池中始终只有core线程在工作。那么当请求太多，core线程处理不过来的时候，用信号量进行阻塞，
保证只有当core线程的某些线程执行完后，阻塞才解开。
这里使用JAVA并发编程一书中的例子来说明信号量的基本用法。
public class BoundedHashSet<T> {
  public static void main(String[] args) throws Exception {
  BoundedHashSet<String> set = new BoundedHashSet<>(2);
  set.add("1");
  set.add("2");
  set.remove("2");
  set.add("3");
  System.out.println(JSON.toJSONString(set));
}
private final Set<T> tempSet;
private final Semaphore sem;
public BoundedHashSet(int size) {
   this.tempSet = Collections.synchronizedSet(new HashSet<T>());
   sem = new Semaphore(size);
}
public boolean add(T o) throws Exception {
　　sem.acquire();
　　boolean isAdd = false;
     try{
     isAdd = tempSet.add(o);
     return isAdd;
}
finally {
  if (!isAdd) {
   sem.release();}}
}
public boolean remove(Object o) {
  boolean isRemoved = tempSet.remove(o);
  if (isRemoved) {
   sem.release();}
  return isRemoved;}
}
这里例子实现了有界阻塞的HashSet。只允许这个HashSet存放两个元素，如果想存第三个元素，必须等到有人把HashSet中的元素remove掉。
每次add之前先申请一个许可，如果能申请到，则正常添加元素。申请不到，则acquire()方法会一直阻塞。remove操作里面，则有一个释放许可的操作。

Semaphore使用事例

21 使用信号量解决死锁问题？

public class BadLockTest {
    protected Object obj1 = new Object();
    protected Object obj2 = new Object();
    protected ExecutorService executorService = Executors.newCachedThreadPool();
    protected Task1 test1=new Task1();
    protected Task2 test2=new Task2();
    public static void main(String[] args) {
      BadLockTest test = new BadLockTest();
      for(int i=0;i<50;i++){
        test.test1.setCount(i);
        test.test2.setCount(i);
        test.executorService.execute(test.test1);
        test.executorService.execute(test.test2);
      } }
    class Task1 implements Runnable {
    public int count;
    public void setCount(int count){
    this.count=count; }
    @Override
    public void run() {
      synchronized (obj1) {
        System.out.println("task1得到obj1对象锁"+count);
        synchronized (obj2) {
        System.out.println("task1得到obj2对象锁"+count);}}
    } }
   class Task2 implements Runnable {
    public int count;
    public void setCount(int count){
      this.count=count;
    }
    @Override
    public void run() {
     synchronized (obj2) {
       System.out.println("task1得到obj1对象锁"+count);
       synchronized (obj1) {
        System.out.println("task1得到obj2对象锁"+count);}}
    }}
}
得到结果:
task1得到obj1对象锁1
task1得到obj1对象锁1

可从结果就知道已经发生了死锁。
信号量可以控制资源能被多少线程访问，这里我们指定只能被一个线程访问，就做到了类似锁住。
而信号量可以指定去获取的超时时间，我们可以根据这个超时时间，去做一个额外处理。
对于无法成功获取的情况，一般就是重复尝试，或指定尝试的次数，也可以马上退出。

public class BadLockTest {
    protected ExecutorService executorService = Executors.newCachedThreadPool();
    protected Task1 test1=new Task1();
    protected Task2 test2=new Task2();
    protected Semaphore s1=new Semaphore(1);
    protected Semaphore s2=new Semaphore(1);
    public static void main(String[] args) {
     BadLockTest test = new BadLockTest();
     for(int i=0;i<50;i++){
      test.test1.setCount(i);
      test.test2.setCount(i);
      test.executorService.execute(test.test1);
      test.executorService.execute(test.test2);}
    }
   class Task1 implements Runnable {
    public int count;
    public void setCount(int count){
    this.count=count; }
    public void run() {
    try {
     if(s2.tryAcquire(1, TimeUnit.SECONDS)){
      System.out.println("task1得到obj1对象锁"+count);
      if(s1.tryAcquire(1, TimeUnit.SECONDS)){
        System.out.println("task1得到obj2对象锁"+count);}
      }
      s2.release();
      s1.release();}
    }}
   class Task2 implements Runnable {
    public int count;
    public void setCount(int count){
    this.count=count; }
    @Override
    public void run() {
    // synchronized (obj2) {
    try {
      if(s1.tryAcquire(1, TimeUnit.SECONDS)){
        System.out.println("task2得到obj1对象锁"+count);
        if(s2.tryAcquire(1, TimeUnit.SECONDS)){
         System.out.println("task2得到obj2对象锁"+count);}
       }
       s1.release();
       s2.release();}
    }}
}
结果:
task1得到obj2对象锁49
task2得到obj2对象锁49

22. 线程之Semaphore生产消费模型

示例：假设当前有5辆车，10个司机，每个司机轮流用车。
public class Semaphore_Test {
    // 同时只能有5个线程访问某个资源车
    private static Semaphore semaphore = new Semaphore(5);
    public static void main(String[] args) {
        // 10个司机
        for (int i = 1; i <= 10; i++) {
            (new Driver(i)).start();
        }
    }
    static class Driver extends Thread {
        private int i;
        public Driver(int i) {
            this.i = i;
        }
        public void run() {
         try {
            semaphore.acquire();
            System.out.println("Driver " + i + " is using car");
            Thread.sleep(1000);
            System.out.println("Driver " + i + " return back car");
            semaphore.release();} 
        }}
}
可能的输出如下：
Driver 1 is using car
Driver 6 is using car
Driver 5 is using car
Driver 2 is using car
Driver 3 is using car
Driver 3 return back car
Driver 5 return back car
Driver 4 is using car
Driver 2 return back car
Driver 1 return back car
Driver 6 return back car
Driver 9 is using car
Driver 8 is using car
Driver 7 is using car
Driver 10 is using car
Driver 4 return back car
Driver 10 return back car
Driver 8 return back car
Driver 7 return back car
Driver 9 return back

3. Semaphore的使用案例(生产和消费事例)？

需求:使用2个线程，分别代表：生产者、消费者。让他们并发的去生产、消费产品，生产的总数是不能超过N的。

实现思路：
这里使用是使用信号量去控制线程的生产消费，通过释放令牌的形式去控制生产者消费者的上限。使用互斥锁保证每次最多只有一个角色去修改共享变量。

/**
 * 生产者消费者模型
 */
public class ProducerConsumerProblem {
    private static final int N = 10; //初始容量
    /***
     * full 产品容量
     * empty 空余容量
     * mutex 读写锁
     */
    private static Semaphore full, empty, mutex;
    //记录当前的产品数量
    private static volatile int count = 0;
    static {
    /**
     * empty 初始化有N个空余位置放置产品
     * mutex 初始化每次最多只有一个线程可以读写
     */
        full = new Semaphore(0);
        empty = new Semaphore(N);
        mutex = new Semaphore(1);
    }
    public static void main(String[] args) {
        new Thread(new Producer()).start();
        new Thread(new Consumer()).start();
    }
    static class Producer implements Runnable {
      @Override
      public void run() {
        while (true) {
         try {
            empty.acquire();//等待空位
            mutex.acquire();//等待读写锁
            count++;
            out.println("生产者生产了一个，还剩：" + count);
            mutex.release();//释放读写锁
            full.release();//放置产品
　　　　　     //随机休息一段时间，让生产者线程有机会抢占读写锁
            Thread.sleep(((int) Math.random()) % 10);
         } }
     }
    }
    static class Consumer implements Runnable {
        @Override
        public void run() {
         while (true) {
          try {
            full.acquire();//等待产品
            mutex.acquire();//等待读写锁
            count--;
            out.println("消费者消费了一个，还剩：" + count);
            mutex.release();//释放读写锁
            empty.release();//释放空位
　　　　　　　          // 随机休息一段时间，让消费者线程有机会抢占读写锁
            Thread.sleep(((int) Math.random()) % 10);
          } }
     }}
}

4.java信号量解决生产消费问题

信号量的基本原理： 简单点理解，信号量可看成一个整形量，表示可用资源的个数,使用资源时信号量-1，释放资源时信号量+1，以生产消费为例:

1. 假设此时缓冲区已满，sem=0,表示当前空的缓冲区个数为0
2. 生产者P1生产, sem=-1,陷入阻塞
3. 生产者P2生产，sem=-2，陷入阻塞
4. 消费者C1消费，sem=-1，唤醒P1
5. 消费者C2消费，sem=0，唤醒P2

归纳：
信号量的值可以反映有多少线程正在等待资源，信号量<0，表示有线程因无资源而陷入等待，如sem=-2,就表示有2个生产者陷入阻塞；
消费时sem++, 执行后若sem<=0, 表示之前有线程阻塞，需要唤醒，如上例中唤醒P1，P2。
信号量个数与初值的选择：
生产消费问题中，需要判断空和满 2个状态。 首先需要2个信号量, empty表示空缓冲区个数，full表示满缓冲区个数，
初值empty=缓冲区大小，full=0, 表示初始状态缓冲区全空，满的缓冲区个数为0。还需要信号量mutex=1保证互斥，初值为1为什么可保持互斥请看P,V操作。
引入P,V操作：
P:表示申请资源。 V:表示释放资源。(P,V操作都是原子性的)

P(semaphore s) {
    s.value--;
    if (s.value < 0)
      当前线程阻塞
}

V(semaphore s) {
    s.value++;
    if (s.value <= 0)
       唤醒等待在该资源上的线程
}

P,V解决生产消费

 semaphore full = 0;
    semaphore empty = BUFF_SIZE;
    semaphore mutex = 1;
    Producer() {
        P(empty);//要生产，则耗费空资源,empty--
        P(mutex);//保证互斥
        生产
.....
        V(mutex);
        V(full);//生产完成则产生1个满资源，full++
    }
    Consumer() {
        P(full);
        P(mutex);
        消费
...
        V(mutex);
        V(empty);
    }

完整的java代码
java.util.concurrent.Semaphore,java中已实现好了Semaphore这个类，可在构造方法中设定初值，require和release方法对应P，V操作

public class SynStack {
    private char[] data = new char[6];
    int cnt=0;
    Semaphore empty = new Semaphore(data.length);//空信号量
    Semaphore full = new Semaphore(0);//满信号量
    Semaphore mutex = new Semaphore(1);//互斥信号量

    //生产
    public void push(char ch){
        try {
            empty.acquire();
            mutex.acquire();
            data[cnt]=ch;
            cnt++;
            System.out.println("生产线程，生产第"+cnt+"个产品，此产品是"+ch);
        }finally{
            mutex.release();
            full.release();
        }
    }
    //消费
    public char pop(){
        char ch=' ';
        try {
            full.acquire();
            mutex.acquire();
            ch=data[cnt-1];
            System.out.println("消费线程，消费第"+cnt+"个产品，此产品是"+ch);
            --cnt;
        }finally{
            mutex.release();
            empty.release();
        }
        return ch;
    }
}
public class Producer implements Runnable{
    private SynStack ss =null;
    public Producer(SynStack ss){
        this.ss = ss;
    }
    @Override
    public void run() {
        char ch;
        for(int i=0;i<15;++i){
            ch =(char)('a'+i);
            ss.push(ch);
    /*try {
        Thread.sleep(1000);
    }*/
    }
    }
}
public class Consumer implements Runnable{
    private SynStack ss = null;
    public Consumer(SynStack ss){
        this.ss = ss;
    }
    @Override
    public void run() {
        for(int i=0;i<15;++i){
            ss.pop();
        }
    }
}
public class Test {
    public static void main(String[] args) {
        SynStack ss = new SynStack();
        Producer producer= new Producer(ss);
        Consumer consumer =new Consumer(ss);
        Thread threadP = new Thread(producer);
        Thread threadC = new Thread(consumer);
        //Thread threadC2= new Thread(consumer);
        threadP.start();
        threadC.start();
        //threadC2.start();
    }
}

 定用一个整数表示信号量，然后有一个增加操作和一个减少操作，如下

  class Semaphore {
        private volatile int permit;
        public Semaphore(int permit) {
            this.permit = permit;
        }
        public acquire() {
            if (permit <= 0) {//等待
            }
　　　　   //执行permit--操作
        }
        public release() {
　　        //执行permit++ 操作
            if（permit > 0){
　　　 　//唤醒
            }
        }
    }

这里有两点需要说明：这里先用一个二进制信号量来等效互斥操作；
由于信号量只能通过0值来进行阻塞和唤醒，所以这里必须使用两个信号量来模拟容器空和容器满两种状态

public class Cache {
    private int cacheSize = 0;
    public Semaphore mutex;
    public Semaphore empty; //保证了容器空的时候（empty的信号量<=0), 消费者等待
    public Semaphore full; //保证了容器满的时候（full的信号量 <= 0），生产者等待
    public Cache(int size) {
        mutex = new Semaphore(1); //二进制信号量，表示互斥锁
        empty = new Semaphore(size);
        full = new Semaphore(0);
    }
    public int getCacheSize() throws InterruptedException {
        return cacheSize;
    }
    public void produce() throws InterruptedException {
        empty.acquire(); // 消耗一个空位
        mutex.acquire();
        cacheSize++;
        System.out.println("生产了一个产品， 当前产品数为" + cacheSize);
        mutex.release();
        full.release(); // 增加了一个产品
    }
    public void consume() throws InterruptedException {
        full.acquire(); // 消耗了一个产品
        mutex.acquire();
        cacheSize--;
        System.out.println("消费了一个产品， 当前产品数为" + cacheSize);
        mutex.release();
        empty.release(); // 增加了一个空位
    }
}
public class Consumer implements Runnable {
    private Cache cache;
    public Consumer(Cache cache) {
        this.cache = cache;
    }
    @Override
    public void run() {
        while (true) {
            try {
              cache.consume();
              Thread.sleep(100);
            }
        }
    }
}
public class Producer implements Runnable {
    private Cache cache;
    public Producer(Cache cache) {
        this.cache = cache;
    }
    @Override
    public void run() {
        while (true) {
            try {
                cache.produce();
                Thread.sleep(100);
            } 
        }
    }
}
public class Main {
   public static void main(String[] args) {
    Cache cache = new Cache(10);
    Producer p = new Producer(cache);
    Consumer c = new Consumer(cache);
    int producerCount = 4, consumerCount = 4;
    for (int i = 0; i < producerCount; i++) {
        new Thread(p).start();
    }
    for (int i = 0; i < consumerCount; i++) {
        new Thread(c).start();
    } }
}

看过了信号量，我们看看信号。 用Java"信号"实现的生产者和消费问题，如下：

public class TestSignal {
    static Monitor monitor = new Monitor();
    static class Producer implements Runnable {
        static int num = 1;
        @Override
        public void run() {
            while (true) {
             try {
                monitor.insert(num);
                System.out.println("生产物品" + num);
                num++;
                Thread.sleep(100);} 
            }
        }
    }
    static class Consumer implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    System.out.println("消费物品" + monitor.remove());
                    Thread.sleep(500);} 
            }
        }
    }
    static class Monitor {//管程，只能有一个线程占用
        private final int capacity = 10;
        private int insertIndex = 0; //仓库中当前可以放置物品的位置
        private int removeIndex = 0; //仓库中当前可以拿走物品的位置
        private final Object[] items = new Object[capacity]; //仓库中的所有物品
        int count = 0; //仓库中的现有物品数
        //向仓库中放置物品
        public synchronized void insert(Object item) throws InterruptedException {
        //当仓库已满时，挂起生产线程
            if (count == capacity) {
                wait();
            }
            items[insertIndex++] = item;
            if (insertIndex == capacity) {
                insertIndex = 0;
            }
            count++;
          //当仓库由空变为不空时，唤起消费线程
            if (count == 1) {
                notify();
            }
        }
        //从仓库中拿走物品
        public synchronized Object remove() throws InterruptedException {
      //当仓库没有物品时，挂起消费线程
            if (count == 0) {
                wait();
            }
            Object item = items[removeIndex++];
            if (removeIndex == capacity) {
                removeIndex = 0;
            }
            count--;
      //当仓库由满变为不满时，唤起生产线程
            if (count == capacity - 1) {
                notify();
            }
            return item;
        }
    }
    public static void main(String[] args) {
        new Thread(new Producer()).start();
        new Thread(new Consumer()).start();
    }
}

可以看到，该例子使用一个Monitor类实现仓库中放置和拿走物品的方法，该类相当于一个管程，只能保证同一时刻只能有一个线程使用该类。
具体实现是采用静态类和synchronized方法，保证当insert()调用时拥有Monitor类的类锁，使得remove()无法获得Monitor类的类锁，
同理，保证当remove()调用时拥有Monitor类的类锁，使得remove()无法被调用。里面实现的巧妙之处在于，当count为capacity时，我们会挂起生产进程，当count从capacity变为capacity - 1时，就会唤醒生产进程加入线程调度队列。
同理，当count为0时，我们会挂起消费进程，当count从0变为1时，就会唤醒消费进程加入线程调度队列。
这种方式的执行结果，与信号量的结果类似，不再列出来。

     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;
            }
        }
    public final boolean hasQueuedPredecessors() {
        Node t = tail; // Read fields in reverse initialization order
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }

  final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    public final void acquireSharedInterruptibly(int arg){if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

    public void release() {
        sync.releaseShared(1);
    }
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
     }    
     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;
            }
        }

class Resource {
    public Object getResource() {
        return new Object();
    }
}
class Pool {
    private static final int MAX_AVAILABLE = 100;
    private final int availableSemaphore;
    private Semaphore available;
    public Pool() {
        this(MAX_AVAILABLE);
    }
    public Pool(int available) {
        this(available, false);
    }
    public Pool(int available, boolean fairMode) {
        this.availableSemaphore = available;
        this.available = new Semaphore(available, fairMode);
        items = new Resource[availableSemaphore];
        for (int i = 0; i < availableSemaphore; i ++) {
            items[i] = new Resource();
        }
        used = new boolean[availableSemaphore];
    }
    public int availableSemaphore() {
        return this.availableSemaphore;
    }
    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;
    protected boolean[] used;
    private synchronized Object getNextAvailableItem() {
      for (int i = 0; i < MAX_AVAILABLE; ++i) {
        if (!used[i]) {
          used[i] = true;
           return items[i];
        }
      }
      return null; // not reached
    }
    private 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;
    }
}

public class SemaphoreDemo {
    public static void main(String ... args) {
        Pool pool = new Pool();
        Object resource = null;
        try {
            Resource r = (Resource) pool.getItem();
            resource = r.getResource();
            //do your biz
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            pool.putItem(resource); //release here
        }
    }
}