阻塞队列和生产者-消费者模式

public class BlockingQueueDemo {

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<CocaCola> queue = new ArrayBlockingQueue<>(100); //容量100的队列
        ExecutorService exec = Executors.newCachedThreadPool();
        for (int i = 0; i < 5; i++) {
            exec.execute(new Producer(queue, exec));
        }
        TimeUnit.SECONDS.sleep(3); //先生产一点库存
        for (int i = 0; i < 5; i++) {
            exec.execute(new FatIndoorsman(queue, exec));
        }
    }
}

class CocaCola { //可口可乐

}

class Producer implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private static List<Producer> producers = new ArrayList<>(); //类管理其实例列表
    private Executor exec;
    private BlockingQueue queue;

    public Producer(BlockingQueue queue, Executor exec) {
        this.queue = queue;
        this.exec = exec;
        producers.add(this);
    }

    public synchronized static void adjust(int flag, BlockingQueue queue, Executor exec) { // 1 添加  -1减少
        if (flag == 1) {
            Producer producer = new Producer(queue, exec); //添加的生产者共享同一个队列
            exec.execute(producer);
        } else if (flag == -1) {
            Producer producer = producers.remove(0);
            producer.cancel();
        }
    }

    private void cancel() { //利用中断取消生产任务
        Thread.currentThread().interrupt();
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                TimeUnit.SECONDS.sleep(1); //模拟生产需耗时1秒
                boolean success = queue.offer(new CocaCola()); //通过offer尝试添加
                if (!success) { //如果队列已满，则移除1个生产者
                    System.out.println("remove a producer");
                    adjust(-1, queue, exec);
                }
                System.out.println(this + " produced a coca-cola!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println(this + " is stoped!");
    }

    @Override
    public String toString() {
        return "Producer[" + id + "]";
    }
}

class FatIndoorsman implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private BlockingQueue queue;
    private Executor exec;

    public FatIndoorsman(BlockingQueue queue, Executor exec) {
        this.queue = queue;
        this.exec = exec;
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            CocaCola cocaCola = (CocaCola) queue.poll();
            if (cocaCola != null) {
                try {
                    TimeUnit.SECONDS.sleep(10); //模拟肥宅每隔10秒要喝一瓶
                    System.out.println(this + " drink a coca-cola");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                Producer.adjust(1, queue, exec); //添加生产者
            }
        }
    }

    @Override
    public String toString() {
        return "FatIndoorsman[" + id + "]";
    }
}