import java.util.Arrays;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * N個(gè)線程向隊(duì)列添加數(shù)據(jù)
 * 一個(gè)線程消費(fèi)隊(duì)列數(shù)據(jù)
 */
public class QueueTest {
  private static List<String> data = Arrays.asList("a", "b", "c", "d", "e");

  private static final int OFFER_COUNT = 40; // 開啟的線程數(shù)量

  private static Semaphore semaphore = new Semaphore(20); // 同一時(shí)間執(zhí)行的線程數(shù)量（大多用于控制API調(diào)用次數(shù)或數(shù)據(jù)庫查詢連接數(shù)）

  public static void main(String[] args) throws InterruptedException {
    Queue<String> queue = new ConcurrentLinkedQueue<>(); // 處理隊(duì)列，需要處理的數(shù)據(jù)，放置到此隊(duì)列中

    CountDownLatch offerLatch = new CountDownLatch(OFFER_COUNT); // offer線程latch，每完成一個(gè)，latch減一，lacth的count為0時(shí)表示offer處理完畢
    CountDownLatch pollLatch = new CountDownLatch(1); // poll線程latch，latch的count為0時(shí)，表示poll處理完畢

    Runnable offerRunnable = () -> {
      try {
        semaphore.acquire(); // 信號(hào)量控制
      } catch (InterruptedException e) {
        e.printStackTrace();
      }

      try {
        for (String datum : data) {
          queue.offer(datum);
          TimeUnit.SECONDS.sleep(2); // 模擬取數(shù)據(jù)很慢的情況
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      } finally {
        // 在finally中執(zhí)行l(wèi)atch.countDown()以及信號(hào)量釋放，避免因異常導(dǎo)致沒有正常釋放
        offerLatch.countDown();
        semaphore.release();
      }
    };

    Runnable pollRunnable = () -> {
      int count = 0;
      try {
        while (offerLatch.getCount() > 0 || queue.size() > 0) { // 只要offer的latch未執(zhí)行完，或queue仍舊有數(shù)據(jù)，則繼續(xù)循環(huán)
          String poll = queue.poll();
          if (poll != null) {
            System.out.println(poll);
            count++;
          }
          // 無論是否poll到數(shù)據(jù)，均暫停一小段時(shí)間，可降低CPU消耗
          TimeUnit.MILLISECONDS.sleep(100);
        }
        System.out.println("total count:" + count);
      } catch (InterruptedException e) {
        e.printStackTrace();
      } finally {
        // 在finally中執(zhí)行l(wèi)atch.countDown()，避免因異常導(dǎo)致沒有正常釋放
        pollLatch.countDown();
      }
    };

    // 啟動(dòng)線程（生產(chǎn)環(huán)境中建議使用線程池）
    new Thread(pollRunnable).start(); // 啟動(dòng)一個(gè)poll線程
    for (int i = 0; i < OFFER_COUNT; i++) {
      new Thread(offerRunnable).start();
    } // 模擬取數(shù)據(jù)很慢，需要開啟40個(gè)線程處理

    // latch等待，會(huì)block主線程直到latch的count為0
    offerLatch.await();
    pollLatch.await();

    System.out.println("===the end===");
  }
}