BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<> (666);

/**
 * 構(gòu)造方法
 * 還有兩個(gè)構(gòu)造函數(shù)，一個(gè)無(wú)fair參數(shù)，一個(gè)可傳入集合，創(chuàng)建時(shí)插入隊(duì)列
 * @param capacity 固定容量
 * @param fair 默認(rèn)是false:訪問(wèn)順序未指定; true:按照FIFO順序處理
 */
public ArrayBlockingQueue(int capacity, boolean fair) {
 ? if (capacity <= 0)
 ? ? ? ?throw new IllegalArgumentException();
 ? ?this.items = new Object[capacity];
 ? ?lock = new ReentrantLock(fair); // 根據(jù)fair創(chuàng)建對(duì)應(yīng)的鎖
 ? ?// 條件對(duì)象，配合容器能滿足業(yè)務(wù)
 ? ?notEmpty = lock.newCondition(); // 出隊(duì)條件對(duì)象
 ? ?notFull = ?lock.newCondition(); // 入隊(duì)條件對(duì)象
}
/**
 * 入隊(duì)方法
 * 在隊(duì)列的尾部插入指定的元素，如果隊(duì)列已滿，則等待空間可用
 */
public void put(E e) throws InterruptedException {
 ? ?checkNotNull(e); // 檢查put對(duì)象是否為空，空拋出異常
 ? ?final ReentrantLock lock = this.lock;
 ? ?lock.lockInterruptibly(); // 若未被中斷嘗試獲取鎖，詳見(jiàn)下文
 ? ?try {
 ?      // 隊(duì)列中元素的數(shù)量 等于 排隊(duì)元素的長(zhǎng)度
 ? ? ? ?while (count == items.length)
 ? ? ? ? ? ?notFull.await(); // 見(jiàn)下文
 ? ? ? ?enqueue(e); // 元素入隊(duì)
 ?  } finally {
 ? ? ? ?lock.unlock();
 ?  }
}
/**
 * 出隊(duì)方法
 * 獲取隊(duì)列的頭部元素并將其刪除，如果隊(duì)列為空，則阻塞并等待元素變?yōu)榭捎?
 */
public E take() throws InterruptedException {
 ? ?final ReentrantLock lock = this.lock;
 ? ?lock.lockInterruptibly(); // 見(jiàn)下文
 ? ?try {
 ? ? ? ?while (count == 0)
 ? ? ? ? ? ?notEmpty.await(); // 見(jiàn)下文
 ? ? ? ?return dequeue(); // 元素出隊(duì)
 ?  } finally {
 ? ? ? ?lock.unlock();
 ?  }
}

@Slf4j
@Slf4j
public class Producer implements Runnable{
 ? ?// 作為參數(shù)的阻塞隊(duì)列
 ? ?private BlockingQueue<Integer> blockingQueue;
 ? ?private final int stopTag;
 ? ?/**
 ? ? * 構(gòu)造方法
 ? ? * @param blockingQueue
 ? ? * @param stopTag
 ? ? */
 ? ?public Producer(BlockingQueue<Integer> blockingQueue,int stopTag) {
 ? ? ? ?this.blockingQueue = blockingQueue;
 ? ? ? ?this.stopTag = stopTag;
 ?  }
 ? ?@Override
 ? ?public void run() {
 ? ? ? ?try {
 ? ? ? ? ? ?generateNumbers();
 ? ? ?  } catch (InterruptedException e) {
 ? ? ? ? ? ?Thread.currentThread().interrupt();
 ? ? ?  }
 ?  }
? ?private void generateNumbers() throws InterruptedException {
 ? ? ? ?// 每個(gè)生產(chǎn)者都隨機(jī)生產(chǎn)10種商品
 ? ? ? ?for (int i = 0; i < 10; i++) {
 ? ? ? ? ? ?int product = ThreadLocalRandom.current().nextInt(1000,1100);
 ? ? ? ? ? ?log.info("生產(chǎn)者{}號(hào),生產(chǎn)了商品,編號(hào)為{}",Thread.currentThread().getId(),product);
 ? ? ? ? ? ?blockingQueue.put(product);
 ? ? ?  }
 ? ? ? ?// 生產(chǎn)終止標(biāo)記
 ? ? ? ?blockingQueue.put(stopTag);
 ? ? ? ?log.info("生產(chǎn)者{}號(hào),生產(chǎn)了第終止標(biāo)記編號(hào){}",Thread.currentThread().getId(),Thread.currentThread().getId());
 ?  }
}

@Slf4j
public class Consumer implements Runnable{
 ? ?// 作為參數(shù)的阻塞隊(duì)列
 ? ?private BlockingQueue<Integer> queue;
 ? ?private final int stopTage;
 ? ?public Consumer(BlockingQueue<Integer> queue, int stopTage) {
 ? ? ? ?this.queue = queue;
 ? ? ? ?this.stopTage = stopTage;
 ?  }
 ? ?@Override
 ? ?public void run() {
 ? ? ? ?try {
 ? ? ? ? ? ?while (true) {
 ? ? ? ? ? ? ? ?Integer product = queue.take();
 ? ? ? ? ? ? ? ?if (product.equals(stopTage)) {
 ? ? ? ? ? ? ? ? ? ?log.info("{}號(hào)消費(fèi)者，停止消費(fèi),因?yàn)槟玫搅送Ｖ瓜M(fèi)標(biāo)記",Thread.currentThread().getId());
 ? ? ? ? ? ? ? ? ? ?return;
 ? ? ? ? ? ? ?  }
 ? ? ? ? ? ? ? ?log.info("{}號(hào)消費(fèi)者,拿到的商品編號(hào):{}",Thread.currentThread().getId(),product);
 ? ? ? ? ?  }
 ? ? ?  } catch (InterruptedException e) {
 ? ? ? ? ? ?Thread.currentThread().interrupt();
 ? ? ?  }
 ?  }
}

public class ProductConsumerTest {
 ? ?public static void main(String[] args) {
 ? ? ? ?// 阻塞隊(duì)列容量
 ? ? ? ?int blockingQueueSize = 10;
 ? ? ? ?// 生產(chǎn)者數(shù)量
 ? ? ? ?int producerSize = 16;
 ? ? ? ?// 消費(fèi)者數(shù)量 = 計(jì)算機(jī)線程核數(shù) 8
 ? ? ? ?int consumerSize = Runtime.getRuntime().availableProcessors();
 ? ? ? ?// 終止消費(fèi)標(biāo)記
 ? ? ? ?int stopTag = Integer.MAX_VALUE;
 ? ? ? ?BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(blockingQueueSize);
 ? ? ? ?// 創(chuàng)建16個(gè)生產(chǎn)者線程
 ? ? ? ?for (int i = 0; i < producerSize; i++) {
 ? ? ? ? ? ?new Thread(new Producer(blockingQueue, stopTag)).start();
 ? ? ?  }
 ? ? ? ?// 創(chuàng)建8個(gè)消費(fèi)者線程
 ? ? ? ?for (int j = 0; j < consumerSize; j++) {
 ? ? ? ? ? ?new Thread(new Consumer(blockingQueue, stopTag)).start();
 ? ? ?  }
 ?  }
}

/**
 * 電影票類(lèi)，實(shí)現(xiàn)了Delayed接口，重寫(xiě) compareTo 和 getDelay方法
 */
public class MovieTicket implements Delayed {
 ? ?//延遲時(shí)間
 ? ?private final long delay;
 ? ?//到期時(shí)間
 ? ?private final long expire;
 ? ?//數(shù)據(jù)
 ? ?private final String msg;
 ? ?//創(chuàng)建時(shí)間
 ? ?private final long now;
 ? ?public long getDelay() {
 ? ? ? ?return delay;
 ?  }
 ? ?public long getExpire() {
 ? ? ? ?return expire;
 ?  }
 ? ?public String getMsg() {
 ? ? ? ?return msg;
 ?  }
 ? ?public long getNow() {
 ? ? ? ?return now;
 ?  }
 ? ?/**
 ? ? * @param msg 消息
 ? ? * @param delay 延期時(shí)間
 ? ? */
 ? ?public MovieTicket(String msg , long delay) {
 ? ? ? ?this.delay = delay;
 ? ? ? ?this.msg = msg;
 ? ? ? ?expire = System.currentTimeMillis() + delay; ? ?//到期時(shí)間 = 當(dāng)前時(shí)間+延遲時(shí)間
 ? ? ? ?now = System.currentTimeMillis();
 ?  }
 ? ?/**
 ? ? * @param msg
 ? ? */
 ? ?public MovieTicket(String msg){
 ? ? ? ?this(msg,1000);
 ?  }
 ? ?public MovieTicket(){
 ? ? ? ?this(null,1000);
 ?  }
 ? ?/**
 ? ? * 獲得延遲時(shí)間 ? 用過(guò)期時(shí)間-當(dāng)前時(shí)間,時(shí)間單位毫秒
 ? ? * @param unit
 ? ? * @return
 ? ? */
 ? ?@Override
 ? ?public long getDelay(TimeUnit unit) {
 ? ? ? ?return unit.convert(this.expire
 ? ? ? ? ? ? ? ?- System.currentTimeMillis() , TimeUnit.MILLISECONDS);
 ?  }
 ? ?/**
 ? ? * 用于延遲隊(duì)列內(nèi)部比較排序  當(dāng)前時(shí)間的延遲時(shí)間 - 比較對(duì)象的延遲時(shí)間
 ? ? * 越早過(guò)期的時(shí)間在隊(duì)列中越靠前
 ? ? * @param delayed
 ? ? * @return
 ? ? */
 ? ?@Override
 ? ?public int compareTo(Delayed delayed) {
 ? ? ? ?return (int) (this.getDelay(TimeUnit.MILLISECONDS)
 ? ? ? ? ? ? ? ?- delayed.getDelay(TimeUnit.MILLISECONDS));
 ?  }
}

public static void main(String[] args) {
 ? ?DelayQueue<MovieTicket> delayQueue = new DelayQueue<MovieTicket>();
 ? ?MovieTicket ticket = new MovieTicket("電影票1",10000);
 ? ?delayQueue.put(ticket);
 ? ?MovieTicket ticket1 = new MovieTicket("電影票2",5000);
 ? ?delayQueue.put(ticket1);
 ? ?MovieTicket ticket2 = new MovieTicket("電影票3",8000);
 ? ?delayQueue.put(ticket2);
 ? ?log.info("message:--->入隊(duì)完畢");
 ? ?while( delayQueue.size() > 0 ){
 ? ? ? ?try {
 ? ? ? ? ? ?ticket = delayQueue.take();
 ? ? ? ? ? ?log.info("電影票出隊(duì):{}",ticket.getMsg());
 ? ? ?  } catch (InterruptedException e) {
 ? ? ? ? ? ?e.printStackTrace();
 ? ? ?  }
 ?  }
}