深入Java多线程和并发编程之Semaphore

以前在学操作系统时就接触过信号量（Semaphore['seməfɔ:]）的概念，指的是多线程环境下保证一个或多个关键代码段不被并发调用。

举一个简单的例子可以帮助理解：比如车库的容量是有限的、同一时刻只能停100辆车、超过了100辆、其他的车辆就只能等待了。

在Jdk的java.util.concurrent包中已经实现了一个Semaphore类、我们可以直接拿来用、不需要再重复制造轮子了。

该类有如下一些特点：
1、Semaphore是带有Counting、相当于维护了一定数量的通行证、如上面举的例子里面的100辆车
2、该类的acquire方法是为了申请通行证的、该方法可能会Block（这里并没有使用synchronization、因为否则的话release也没法执行）、就是当通行证全部被申请完了之后。而对应的release方法则是释放通行证的、这样该方法执行后可能会让原来Block的acquire方法得以继续执行。
3、上面2的执行过程中其实并没有类似于通行证这样的对象、Semaphore只是维护了一个许可的计数器。
4、如果Semaphore的计数器为1（又叫binary Semaphore）的话则相当于一个互斥锁

实际应用中、对象池和线程池都有使用如此的方式控制线程的使用量。


下面是一个代码示例：

package com.alibaba.plouto.test.mock;

import java.util.concurrent.Semaphore;

public class TestSemaphore {
    public static CarWarehouse carWarehouse = new CarWarehouse();

    public static void main(String args[]) {
        for (int i = 0; i <= 10; i++) {
            new Thread(new Producer(i)).start();
            new Thread(new Consumer()).start();
        }
    }

    static class CarWarehouse {
        final int       MAX_PERMITS = 5;
        final Semaphore permits     = new Semaphore(MAX_PERMITS);
        final Semaphore mutex       = new Semaphore(1);
        // 库存容量  
        final Object[]  carWaits    = new Object[MAX_PERMITS];
        int             count;

        public void put(Object x) throws InterruptedException {
            permits.acquire();
            for (int i = 0; i < MAX_PERMITS; i++) {
                if (carWaits[i] == null) {
                    //防止两辆车正好进同一个车位（当车库没满时、可以同时进多辆车）
                    mutex.acquire();
                    //通过Double Check防止细粒度的并发问题
                    if (carWaits[i] == null) {
                        carWaits[i] = x;
                        System.out.println("put " + carWaits[i]);
                        //停到了车位、释放互斥锁
                        mutex.release();
                        break;
                    } else {
                        //发现该车位满了、也需要释放互斥锁、继续寻找下一个位置
                        mutex.release();
                    }
                }
            }
        }

        public void take() throws InterruptedException {
            int i = 0;
            while (true) {
                if (carWaits[i] != null) {
                    mutex.acquire();
                    if (carWaits[i] != null) {
                        System.out.println("take " + carWaits[i]);
                        carWaits[i] = null;
                        permits.release();
                        mutex.release();
                        break;
                    } else {
                        mutex.release();
                    }
                }
                i++;
                if (i == MAX_PERMITS)
                    i = 0;
            }
        }
    }

    static class Producer implements Runnable {
        private String car;

        public Producer(int i) {
            car = "car" + i;
        }

        public void run() {
            try {
                carWarehouse.put(car);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Consumer implements Runnable {

        public void run() {
            try {
                carWarehouse.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}