Semaphore简介

Semaphore 作用

Semaphore 通常叫它信号量，可以用来控制同时访问特定资源的线程数量，通过协调各个线程，以保证合理的使用资源。可以把它简单的理解成停车场入口立着的那个显示屏，每有一辆车进入停车场显示屏就会显示剩余车位减1，每有一辆车从停车场出去，显示屏上显示的剩余车辆就会加1，当显示屏上的剩余车位为0时，停车场入口的栏杆就不会再打开，车辆就无法进入停车场了，直到有一辆车从停车场出去为止。

可以看到，Semaphore 可以达到一个限流线程数的效果。

常用方法

//获取一个令牌，在获取到令牌、或者被其他线程调用中断之前线程一直处于阻塞状态。
acquire()  
//获取一个令牌，在获取到令牌、或者被其他线程调用中断、或超时之前线程一直处于阻塞状态。
acquire(int permits)  
//获取一个令牌，在获取到令牌之前线程一直处于阻塞状态（忽略中断）。
acquireUninterruptibly() 
//尝试获得令牌，返回获取令牌成功或失败，不阻塞线程。
tryAcquire()
//尝试获得令牌，在超时时间内循环尝试获取，直到尝试获取成功或超时返回，不阻塞线程。
tryAcquire(long timeout, TimeUnit unit)
//释放一个令牌，唤醒一个获取令牌不成功的阻塞线程。    
release()
//等待队列里是否还存在等待线程。
hasQueuedThreads()
//获取等待队列里阻塞的线程数。
getQueueLength()
//清空令牌把可用令牌数置为0，返回清空令牌的数量。 
drainPermits()
//返回可用的令牌数量。
availablePermits()

代码演示

@Slf4j(topic = "c.main")
public class Main {
    public static void main(String[] args) {
        // 1. 创建 semaphore 对象
        Semaphore semaphore = new Semaphore(3);
        // 2. 10个线程同时运行
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                // 3. 获取许可
                try {
                    semaphore.acquire();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    log.debug("running...");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    log.debug("end...");
                } finally {
                    // 4. 释放许可
                    semaphore.release();
                }
            }).start();
        }
    }
}

测试结果：

可以看到明显将线程数限制在了3个。

实例应用

使用场景

数据库连接池：同时进行连接的线程有数量限制，连接不能超过一定的数量，当连接达到了限制数量后，后面的线程只能排队等前面的线程释放了数据库连接才能获得数据库连接。

限流线程数：在访问高峰期时，让请求线程阻塞，高峰期过去再释放许可，当然它只适合限制单机线程数量，并且仅是限制线程数，而不是限制资源数（例如连接数，请对比 Tomcat LimitLatch 的实现）

代码示例

实现数据库连接池等待机制的代码如下：

@Slf4j(topic = "c.pool")
public class Pool {
  // 1. 连接池大小
  private final int poolSize;
  // 2. 连接对象数组
  private final Connection[] connections;
  // 3. 连接状态数组 0 表示空闲， 1 表示繁忙
  private final AtomicIntegerArray states;
  private final Semaphore semaphore;
  // 4. 构造方法初始化
  public Pool(int poolSize) {
      this.poolSize = poolSize;
      // 许可数与资源数一致
      this.semaphore = new Semaphore(poolSize);
      this.connections = new Connection[poolSize];
      this.states = new AtomicIntegerArray(new int[poolSize]);
      for (int i = 0; i < poolSize; i++) {
          connections[i] = new Connection() {
              //省略各种实现方法
          };
      }
  }
  // 5. 借连接
  public Connection borrow() {// t1, t2, t3
      // 获取许可
      try {
          semaphore.acquire(); // 没有许可的线程，在此一直等待
      } catch (InterruptedException e) {
          e.printStackTrace();
      }
      for (int i = 0; i < poolSize; i++) {
          // 获取空闲连接
          if(states.get(i) == 0) {
              if (states.compareAndSet(i, 0, 1)) {
                  log.debug("borrow {}", connections[i]);
                  return connections[i];
              }
          }
      }
      // 不会执行到这里
      return null;
  }
  // 6. 归还连接
  public void free(Connection conn) {
      for (int i = 0; i < poolSize; i++) {
          if (connections[i] == conn) {
              states.set(i, 0);
              log.debug("free {}", conn);
              semaphore.release();
              break;
          }
      }
  }
}

Java 并发编程

信号量Semaphore