并发编程系列-Semaphore

Semaphore，如今通常被翻译为"信号量"，过去也曾被翻译为"信号灯"，因为类似于现实生活中的红绿灯，车辆是否能通行取决于是否是绿灯。同样，在编程世界中，线程是否能执行取决于信号量是否允许。

信号量是由著名的计算机科学家迪杰斯特拉（Dijkstra）于1965年提出的，直到1980年管程被提出，它一直是并发编程领域的主导方法。如今几乎所有支持并发编程的语言都支持信号量机制，因此掌握信号量仍然非常必要。

下面我们首先介绍信号量模型，然后介绍如何运用信号量，最后使用信号量来实现一个流量控制器。

信号量模型

信号量模型还是很简单的，可以简单概括为： 一个计数器，一个等待队列，三个方法。在信号量模型里，计数器和等待队列对外是透明的，所以只能通过信号量模型提供的三个方法来访问它们，这三个方法分别是：init()、down()和up()。你可以结合下图来形象化地理解。

这三个操作的具体意义如下所示。

初始化（init()）：设定计数器的起始值。
减少（down()）：将计数器的值减1；如果此时计数器的值小于0，则当前线程会被阻塞，否则当前线程可以继续执行。
增加（up()）：将计数器的值加1；如果此时计数器的值小于或等于0，则唤醒等待队列中的一个线程，并将其从等待队列中移除。

上述提到的init()、down()和up()三个操作都是原子的，并且这种原子性由信号量模型的实现方保证。在Java SDK中，信号量模型由java.util.concurrent.Semaphore类实现，Semaphore类能够确保这三个操作都是原子操作。

如果你觉得上面的描述有些复杂，可以参考下面的代码来理解信号量模型的实现。

class Semaphore{
  // 计数器
  int count;
  // 等待队列
  Queue queue;
  // 初始化操作
  Semaphore(int c){
    this.count=c;
  }
  //
  void down(){
    this.count--;
    if(this.count<0){
      //将当前线程插入等待队列
      //阻塞当前线程
    }
  }
  void up(){
    this.count++;
    if(this.count<=0) {
      //移除等待队列中的某个线程T
      //唤醒线程T
    }
  }
}

这里再插一句，信号量模型里面，down()、up()这两个操作历史上最早称为P操作和V操作，所以信号量模型也被称为PV原语。另外，还有些人喜欢用semWait()和semSignal()来称呼它们，虽然叫法不同，但是语义都是相同的。在Java SDK并发包里，down()和up()对应的则是acquire()和release()。

如何使用信号量

通过前文，你应该会发现信号量的模型还是非常简单的，那具体应该如何使用呢？其实你可以想象一下红绿灯。在十字路口，红绿灯可以控制交通流量，这得益于一个重要的规则：车辆在通过路口前必须检查是否为绿灯，只有绿灯才能通行。这个规则与我们之前提到的锁规则很相似，不是吗？

实际上，信号量的使用也是类似的。让我们继续用累加器的例子来说明信号量的使用吧。在累加器的例子中，count += 1操作是一个临界区，只允许一个线程执行，也就是说需要保证互斥性。那么，在这种情况下，如何使用信号量来控制呢？

实际上非常简单，就像我们使用互斥锁一样，只需要在进入临界区之前执行一次down()操作，在退出临界区之前执行一次up()操作就可以了。下面是Java代码的示例，acquire()就是信号量中的down()操作，release()就是信号量中的up()操作。

static int count;
//初始化信号量
static final Semaphore s
    = new Semaphore(1);
//用信号量保证互斥
static void addOne() {
  s.acquire();
  try {
    count+=1;
  } finally {
    s.release();
  }
}

接下来我们再来分析一下，信号量是如何确保互斥性的。假设有两个线程T1和T2同时访问addOne()方法，当它们同时调用acquire()时，由于acquire()是一个原子操作，因此只能有一个线程（假设是T1）将信号量中的计数器减为0，而另一个线程（T2）将计数器减为-1。对于线程T1来说，信号量中计数器的值是0，大于等于0，所以线程T1会继续执行；对于线程T2来说，信号量中计数器的值是-1，小于0，根据信号量模型中对down()操作的描述，线程T2将被阻塞。因此，此时只有线程T1可以进入临界区并执行 count += 1；。

当线程T1执行release()操作，也就是up()操作时，信号量中计数器的值是-1，经过加1后的值是0，小于等于0，根据信号量模型中对up()操作的描述，此时等待队列中的T2将被唤醒。于是，在T1执行完临界区代码之后，T2才有机会进入临界区执行，从而确保了互斥性。

快速实现一个限流器

上述的示例，我们通过使用信号量实现了一个简单的互斥锁功能。也许你会觉得奇怪，既然Java的SDK中已经提供了Lock，为什么还需要提供Semaphore呢？实际上，互斥锁只是Semaphore的一部分功能，而Semaphore还有一个Lock无法实现的功能，那就是：允许多个线程访问临界区。

实际情况中确实存在这样的需求。比较常见的例子是我们在工作中遇到的各种资源池，比如连接池、对象池、线程池等等。其中，你可能对数据库连接池最为熟悉，在同一时刻，允许多个线程同时使用连接池，但每个连接在释放之前不允许其他线程使用。

实际上，不久前我在工作中也遇到了一个对象池的需求。对象池指的是一次性创建N个对象，然后所有线程重复利用这些对象，当然在对象释放之前不允许其他线程使用。对象池可以使用List来保存实例对象，这很简单。但关键在于限流器的设计，这里的限流指的是不允许超过N个线程同时进入临界区。那么如何快速实现这样的限流器呢？我立刻想到了使用信号量的解决方案。

在上面的例子中，信号量的计数器被设置为1，这个1表示只允许一个线程进入临界区。但是，如果我们将计数器的值设为对象池中对象的数量N，就可以完美解决对象池的限流问题了。下面是一个对象池的示例代码。

class ObjPool<T, R> {
  final List<T> pool;
  // 用信号量实现限流器
  final Semaphore sem;
  // 构造函数
  ObjPool(int size, T t){
    pool = new Vector<T>(){};
    for(int i=0; i<size; i++){
      pool.add(t);
    }
    sem = new Semaphore(size);
  }
  // 利用对象池的对象，调用func
  R exec(Function<T,R> func) {
    T t = null;
    sem.acquire();
    try {
      t = pool.remove(0);
      return func.apply(t);
    } finally {
      pool.add(t);
      sem.release();
    }
  }
}
// 创建对象池
ObjPool<Long, String> pool =
  new ObjPool<Long, String>(10, 2);
// 通过对象池获取t，之后执行
pool.exec(t -> {
    System.out.println(t);
    return t.toString();
});

我们使用一个List来存储对象实例，并使用Semaphore实现限流器。关键代码位于ObjPool的exec()方法中，这个方法实现了限流的功能。在这个方法中，我们首先调用acquire()方法（在finally块中会调用release()方法），假设对象池大小为10，信号量的计数器初始化为10，那么前10个线程调用acquire()方法后可以继续执行，相当于通过了信号灯，而其他线程将阻塞在acquire()方法上。通过了信号灯的线程，我们为每个线程分配一个对象t（通过pool.remove(0)实现），并执行一个回调函数func，该回调函数的参数正是之前分配的对象t；在执行完回调函数后，它们会释放对象（通过pool.add(t)实现），同时调用release()方法来更新信号量的计数器。如果此时信号量的计数器值小于等于0，说明有线程在等待，此时会自动唤醒等待的线程。

简而言之，使用信号量，我们可以轻松实现一个限流器，且使用起来非常简单。

总结

信号量在Java语言中的知名度相对较低，但在其他编程语言中却非常有名。Java在并发编程领域取得了很快的发展，重点支持的是管程模型。管程模型在理论上解决了信号量模型的一些不足之处，主要体现在易用性和工程化方面。例如，使用信号量来解决我们前面提到的阻塞队列问题比使用管程模型要复杂许多。如果你感兴趣的话，可以进行了解和尝试。

本文由 mdnice 多平台发布