public class BlockingQueue {
    public static void main(String[] args) throws InterruptedException {
        SynchronousQueue queue = new SynchronousQueue();
        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName() + ": put 1");
                queue.put("a");
                System.out.println(Thread.currentThread().getName() + ": put 2");
                queue.put("b");
                System.out.println(Thread.currentThread().getName() + ": put 3");
                queue.put("c");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"AA").start();

        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println("take:" + queue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println("take:" +queue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println("take:" +queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"BB").start();
    }
}

class Source {
    private volatile boolean flag = true; //默认开启消费标记
    private AtomicInteger atomicInteger = new AtomicInteger();

    BlockingQueue<String> queue = null;

    public Source(BlockingQueue<String> blockingQueue) {
        this.queue = blockingQueue;
        System.out.println(queue.getClass().getName());
    }


    //生产
    public void producer() throws Exception {
        String data = null;
        boolean retVal;
        while (flag) {
            data = atomicInteger.incrementAndGet() + "";
            retVal = queue.offer(data, 2L, TimeUnit.SECONDS);
            //插入成功
            if (retVal) {
                System.out.println(Thread.currentThread().getName() + "\t插入数据" + data + "成功");
            } else {
                System.out.println(Thread.currentThread().getName() + "\t插入数据" + data + "失败");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println("生产停止！");
    }

    //消费
    public void consumer() throws Exception {
        String result;
        while (flag) {
            result = queue.poll(2L, TimeUnit.SECONDS);
            if (null == result || result.equalsIgnoreCase("")) {
                System.out.println("队列超过2S无数据！消费退出！");
                flag = false;
                return;
            }
            System.out.println(Thread.currentThread().getName() + "\t消费队列-->" + result + "成功");
        }
    }
}

public class ProducerConsumer {

    public static void main(String[] args) {
        Source source = new Source(new ArrayBlockingQueue<>(2));
        new Thread(() -> {
            try {
                source.producer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "pord").start();


        new Thread(() -> {
            try {
                source.consumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "cons").start();
    }
}