package com.example.demo.study;

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

public class Study0217 {
  //創(chuàng)建一個(gè)信號(hào)量的實(shí)例，信號(hào)量初始值為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");
      //信號(hào)量加一
      semaphore.release();
    });
    
    pool.submit(()->{
      System.out.println("Thread2---start");
      //信號(hào)量加一
      semaphore.release();
    });
    pool.submit(()->{
      System.out.println("Thread3---start");
      //信號(hào)量加一
      semaphore.release();
    });
    //等待兩個(gè)子線程執(zhí)行完畢就放過(guò)，必須要信號(hào)量等于2才放過(guò)
    semaphore.acquire(2);
    System.out.println("兩個(gè)子線程執(zhí)行完畢");
    
    //關(guān)閉線程池，正在執(zhí)行的任務(wù)繼續(xù)執(zhí)行
    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 {
  //創(chuàng)建一個(gè)信號(hào)量的實(shí)例，信號(hào)量初始值為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");
      //信號(hào)量加一
      semaphore.release();
    });
    
    pool.submit(()->{
      System.out.println("Thread2---start");
      //信號(hào)量加一
      semaphore.release();
    });
    
    //等待兩個(gè)子線程執(zhí)行完畢就放過(guò)，必須要信號(hào)量等于2才放過(guò)
    semaphore.acquire(2);
    System.out.println("子線程1，2執(zhí)行完畢");
    
    pool.submit(()->{
      System.out.println("Thread3---start");
      //信號(hào)量加一
      semaphore.release();
    });
    pool.submit(()->{
      System.out.println("Thread4---start");
      //信號(hào)量加一
      semaphore.release();
    });
    
    semaphore.acquire(2);
    System.out.println("子線程3，4執(zhí)行完畢");
    
    //關(guān)閉線程池，正在執(zhí)行的任務(wù)繼續(xù)執(zhí)行
    pool.shutdown();

  }

}

//默認(rèn)是非公平策略
public Semaphore(int permits) {
  sync = new NonfairSync(permits);
}
//可以根據(jù)第二個(gè)參數(shù)選擇是公平策略還是非公平策略
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();
  //這里根據(jù)子類是公平策略還是非公平策略
  if (tryAcquireShared(arg) < 0)
    //獲取失敗會(huì)進(jìn)入這里，將線程放入阻塞隊(duì)列，然后再嘗試，還是失敗的話就調(diào)用park方法掛起當(dāng)前線程
    doAcquireSharedInterruptibly(arg);
}
//非公平策略
protected int tryAcquireShared(int acquires) {
  return nonfairTryAcquireShared(acquires);
}
final int nonfairTryAcquireShared(int acquires) {
  //一個(gè)無(wú)限循環(huán)，獲取state剩余的信號(hào)量，因?yàn)槊空{(diào)用一次release()方法的話，信號(hào)量就會(huì)加一，這里將
  //最新的信號(hào)量減去傳進(jìn)來(lái)的參數(shù)比較，比如有兩個(gè)線程，其中一個(gè)線程已經(jīng)調(diào)用了release方法，然后調(diào)用acquire(2)方法，那么
  //這里remaining的值就是-1,返回-1，然后當(dāng)前線程就會(huì)被丟到阻塞隊(duì)列中去了；如果另外一個(gè)線程也調(diào)用了release方法，
  //那么此時(shí)的remaining==0，所以在這里的if中會(huì)調(diào)用CAS將0設(shè)置到state
  //
  for (;;) {
    int available = getState();
    int remaining = available - acquires;
    if (remaining < 0 || compareAndSetState(available, remaining))
      return remaining;
  }
}
//公平策略
//和上面非公平差不多，只不過(guò)這里會(huì)查看阻塞隊(duì)列中當(dāng)前節(jié)點(diǎn)前面有沒(méi)有前驅(qū)節(jié)點(diǎn)，有的話直接返回-1，
//就會(huì)把當(dāng)前線程丟到阻塞隊(duì)列中阻塞去了,沒(méi)有前驅(qū)節(jié)點(diǎn)的話，就跟非公平模式一樣的了
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;
  }
}

//這個(gè)方法的作用就是將信號(hào)量加一
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)) {
    //釋放資源成功就調(diào)用park方法喚醒喚醒AQS隊(duì)列中最前面的節(jié)點(diǎn)中的線程
    doReleaseShared();
    return true;
  }
  return false;
}

protected final boolean tryReleaseShared(int releases) {
  //一個(gè)無(wú)限循環(huán)，獲取state，然后加上傳進(jìn)去的參數(shù)，如果新的state的值小于舊的state，說(shuō)明已經(jīng)超過(guò)了state的最大值，溢出了
  //沒(méi)有溢出的話，就用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表示節(jié)點(diǎn)中線程需要被喚醒
      if (ws == Node.SIGNAL) {
        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
          continue;      // loop to recheck cases
        //調(diào)用阻塞隊(duì)列中線程的unpark方法喚醒線程
        unparkSuccessor(h);
      }
      //ws == 0表示節(jié)點(diǎn)中線程是初始狀態(tài)
      else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
        continue;        // loop on failed CAS
    }
    
    if (h == head)          // loop if head changed
      break;
  }
}