Java多线程编程实战之模拟大量数据同步

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); // 同一时间执行的线程数量（大多用于控制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中执行latch.countDown()以及信号量释放，避免因异常导致没有正常释放
        offerLatch.countDown();
        semaphore.release();
      }
    };

    Runnable pollRunnable = () -> {
      int count = 0;
      try {
        while (offerLatch.getCount() > 0 || queue.size() > 0) { // 只要offer的latch未执行完，或queue仍旧有数据，则继续循环
          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中执行latch.countDown()，避免因异常导致没有正常释放
        pollLatch.countDown();
      }
    };

    // 启动线程（生产环境中建议使用线程池）
    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===");
  }
}