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個線程向隊列添加數據
 * 一個線程消費隊列數據
 */
public class QueueTest {
  private static List<String> data = Arrays.asList("a", "b", "c", "d", "e");

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

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

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

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

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

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

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

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

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

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