线程通信之生产者消费者阻塞队列版

class MyResource {    private volatile boolean FLAG = true; //默认开启，进行生产+消费    private AtomicInteger atomicInteger = new AtomicInteger();

    BlockingQueue<String> blockingQueue = null;

    public MyResource(BlockingQueue<String> blockingQueue) {        this.blockingQueue = blockingQueue;    }

    public void myProduct() throws Exception {        String data = null;        boolean retValue;        while (FLAG) {            data  = atomicInteger.incrementAndGet()+"";            retValue = blockingQueue.offer(data,2L, TimeUnit.SECONDS);            if(retValue) {                System.out.println(Thread.currentThread().getName() +"\t 插入队列" + data +"成功");            } else {                System.out.println(Thread.currentThread().getName() +"\t 插入队列" + data +"失败");            }            TimeUnit.SECONDS.sleep(1);        }        System.out.println(Thread.currentThread().getName()+"\t 大老板叫停了，表示FLAG = false,生产动作结束");    }

    public void myConsumer() throws Exception {        String result = null;        while(FLAG) {            result = blockingQueue.poll(2L,TimeUnit.SECONDS);            if(null == result || result.equalsIgnoreCase("")) {                FLAG = false;                System.out.println(Thread.currentThread().getName()+"\t超过2秒钟没有取到蛋糕，消费退出");                System.out.println();                System.out.println();                return ;            }            System.out.println(Thread.currentThread().getName()+"\t消费队列"+result +"成功");

        }    }

    public void stop() {        this.FLAG = false;    }}

public class ProductConsumer_BlockQueueDemo {    public static void main(String[] args){        MyResource myResource = new MyResource(new ArrayBlockingQueue<String>(10));        new Thread(() ->{            System.out.println(Thread.currentThread().getName()+"\t生产线程启动");            try {                myResource.myProduct();            } catch (Exception e) {                e.printStackTrace();            }

        },"product").start();

        new Thread(() ->{            System.out.println(Thread.currentThread().getName()+"\t消费线程启动");            try {                myResource.myConsumer();            } catch (Exception e) {                e.printStackTrace();            }

        },"consumer").start();        //暂停一会儿线程        try {TimeUnit.SECONDS.sleep(5);} catch (InterruptedException e) { e.printStackTrace(); }        System.out.println();        System.out.println();        System.out.println();        System.out.println("5秒时间到，大老板main线程叫停，活动结束");        myResource.stop();    }}

运行结果见下图：

原文地址：https://www.cnblogs.com/liuyi13535496566/p/12154242.html