详解Java 信号量Semaphore

package com.example.demo.study;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class Study0217 {
  //创建一个信号量的实例，信号量初始值为0
  static Semaphore semaphore = new Semaphore(0);
  
  public static void main(String[] args) throws Exception {
    ExecutorService pool = Executors.newFixedThreadPool(3);
    pool.submit(()->{
      System.out.println("Thread1---start");
      //信号量加一
      semaphore.release();
    });
    
    pool.submit(()->{
      System.out.println("Thread2---start");
      //信号量加一
      semaphore.release();
    });
    pool.submit(()->{
      System.out.println("Thread3---start");
      //信号量加一
      semaphore.release();
    });
    //等待两个子线程执行完毕就放过，必须要信号量等于2才放过
    semaphore.acquire(2);
    System.out.println("两个子线程执行完毕");
    
    //关闭线程池，正在执行的任务继续执行
    pool.shutdown();

  }

}

package com.example.demo.study;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class Study0217 {
  //创建一个信号量的实例，信号量初始值为0
  static Semaphore semaphore = new Semaphore(0);
  
  public static void main(String[] args) throws Exception {
    ExecutorService pool = Executors.newFixedThreadPool(3);
    pool.submit(()->{
      System.out.println("Thread1---start");
      //信号量加一
      semaphore.release();
    });
    
    pool.submit(()->{
      System.out.println("Thread2---start");
      //信号量加一
      semaphore.release();
    });
    
    //等待两个子线程执行完毕就放过，必须要信号量等于2才放过
    semaphore.acquire(2);
    System.out.println("子线程1，2执行完毕");
    
    pool.submit(()->{
      System.out.println("Thread3---start");
      //信号量加一
      semaphore.release();
    });
    pool.submit(()->{
      System.out.println("Thread4---start");
      //信号量加一
      semaphore.release();
    });
    
    semaphore.acquire(2);
    System.out.println("子线程3，4执行完毕");
    
    //关闭线程池，正在执行的任务继续执行
    pool.shutdown();

  }

}

//默认是非公平策略
public Semaphore(int permits) {
  sync = new NonfairSync(permits);
}
//可以根据第二个参数选择是公平策略还是非公平策略
public Semaphore(int permits, boolean fair) {
  sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

public void acquire(int permits) throws InterruptedException {
  if (permits < 0) throw new IllegalArgumentException();
  sync.acquireSharedInterruptibly(permits);
}

//AQS中的方法
public final void acquireSharedInterruptibly(int arg)
    throws InterruptedException {
  if (Thread.interrupted()) throw new InterruptedException();
  //这里根据子类是公平策略还是非公平策略
  if (tryAcquireShared(arg) < 0)
    //获取失败会进入这里，将线程放入阻塞队列，然后再尝试，还是失败的话就调用park方法挂起当前线程
    doAcquireSharedInterruptibly(arg);
}
//非公平策略
protected int tryAcquireShared(int acquires) {
  return nonfairTryAcquireShared(acquires);
}
final int nonfairTryAcquireShared(int acquires) {
  //一个无限循环，获取state剩余的信号量，因为每调用一次release()方法的话，信号量就会加一，这里将
  //最新的信号量减去传进来的参数比较，比如有两个线程，其中一个线程已经调用了release方法，然后调用acquire(2)方法，那么
  //这里remaining的值就是-1,返回-1，然后当前线程就会被丢到阻塞队列中去了；如果另外一个线程也调用了release方法，
  //那么此时的remaining==0，所以在这里的if中会调用CAS将0设置到state
  //
  for (;;) {
    int available = getState();
    int remaining = available - acquires;
    if (remaining < 0 || compareAndSetState(available, remaining))
      return remaining;
  }
}
//公平策略
//和上面非公平差不多，只不过这里会查看阻塞队列中当前节点前面有没有前驱节点，有的话直接返回-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 void release(int permits) {
  if (permits < 0) throw new IllegalArgumentException();
  sync.releaseShared(permits);
}
//AQS中方法
public final boolean releaseShared(int arg) {
  //tryReleaseShared尝试释放资源
  if (tryReleaseShared(arg)) {
    //释放资源成功就调用park方法唤醒唤醒AQS队列中最前面的节点中的线程
    doReleaseShared();
    return true;
  }
  return false;
}

protected final boolean tryReleaseShared(int releases) {
  //一个无限循环，获取state，然后加上传进去的参数，如果新的state的值小于旧的state，说明已经超过了state的最大值，溢出了
  //没有溢出的话，就用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;
  }
}

private void doReleaseShared() {
  
  for (;;) {
    Node h = head;
    if (h != null && h != tail) {
      int ws = h.waitStatus;
      //ws==Node.SIGNAL表示节点中线程需要被唤醒
      if (ws == Node.SIGNAL) {
        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
          continue;      // loop to recheck cases
        //调用阻塞队列中线程的unpark方法唤醒线程
        unparkSuccessor(h);
      }
      //ws == 0表示节点中线程是初始状态
      else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
        continue;        // loop on failed CAS
    }
    
    if (h == head)          // loop if head changed
      break;
  }
}