java通过信号量实现限流的示例

public class LimitByAtomic {
​
  private static final AtomicInteger COUNTER = new AtomicInteger(10);
​
  public void f() {
    int count = COUNTER.decrementAndGet();
    if (count < 0) {
      COUNTER.incrementAndGet();
      System.out.println("拒绝执行业务逻辑");
      return; // 拒绝执行业务逻辑
    }
​
    try {
      // 执行业务逻辑
      System.out.println("执行业务逻辑");
    } finally {
      COUNTER.incrementAndGet();
    }
  }
}

public class LimitByLock {
​
  private int count = 10;
​
  public void f() {
    if (count <= 0) {
      System.out.println("拒绝执行业务逻辑");
      return;
    }
​
    synchronized (this) {
      if (count <= 0) {
        System.out.println("拒绝执行业务逻辑");
        return;
      }
      count--;
    }
​
    try {
      // 执行业务逻辑
      System.out.println("执行业务逻辑");
    } finally {
      synchronized (this) {
        count++;
      }
    }
  }
}

public class LimitByCondition {
​
  private int count = 10;
​
  public void f() throws Exception {
    synchronized (this) {
      while (count <= 0) {
        System.out.println("等待执行业务逻辑");
        this.wait();
      }
      count--;
    }
​
    try {
      System.out.println("执行业务逻辑");
    } finally {
      synchronized (this) {
        count++;
        this.notifyAll();
      }
    }
  }
}

public class LimitBySemaphore {
​
  private final Semaphore semaphore = new Semaphore(10);
​
  public void f() throws Exception {
    semaphore.acquire();
    try {
      System.out.println("执行业务逻辑");
    } finally {
      semaphore.release();
    }
  }
}

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

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

public class Semaphore implements java.io.Serializable {
  public void acquire() throws InterruptedException;
  public void acquireUninterruptibly();
  public boolean tryAcquire();
  public boolean tryAcquire(long timeout, TimeUnit unit);
  public void release();
  public void acquire(int permits) throws InterruptedException;
  public void acquireUninterruptibly(int permits) ;
  public boolean tryAcquire(int permits);
  public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
    throws InterruptedException;
  public void release(int permits);
}

public class Semaphore implements java.io.Serializable {
    abstract static class Sync extends AbstractQueuedSynchronizer {
    }
​
    // 非公平锁
    static final class NonfairSync extends Sync {
    }
    // 公平锁
    static final class FairSync extends Sync {
    }
    // 默认使用非公平锁
    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
​
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }
}

// java.util.concurrent.Semaphore#acquire()
public void acquire() throws InterruptedException {
  sync.acquireSharedInterruptibly(1);
}
​
// java.util.concurrent.locks.AbstractQueuedSynchronizer#acquireSharedInterruptibly
public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
  // 先判断线程有没有被中断
  if (Thread.interrupted())
    throw new InterruptedException();
  // 尝试获取共享锁，如果获取许可失败，返回值<0, 需要进入等待队列
  if (tryAcquireShared(arg) < 0)
    doAcquireSharedInterruptibly(arg); // 排队等待队列
}
​

// java.util.concurrent.Semaphore.FairSync#tryAcquireShared
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;
  }
}
​
// java.util.concurrent.Semaphore.Sync#nonfairTryAcquireShared
final int nonfairTryAcquireShared(int acquires) {
  for (;;) {
    int available = getState();
    int remaining = available - acquires;
    if (remaining < 0 ||
        compareAndSetState(available, remaining))
      return remaining;
  }
}

// java.util.concurrent.Semaphore#release()
public void release() {
  sync.releaseShared(1);
}
​
// java.util.concurrent.locks.AbstractQueuedSynchronizer#releaseShared
public final boolean releaseShared(int arg) {
  // 尝试释放许可
  if (tryReleaseShared(arg)) {
    // 唤醒等待队列其中一个线程
    doReleaseShared();
    return true;
  }
  return false;
}
​
// java.util.concurrent.Semaphore.Sync#tryReleaseShared
protected final boolean tryReleaseShared(int releases) {
  // 采用自旋 + CAS来更新state
  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 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();
});
​