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

Posted 2021-03-18 liuyi13535496566

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() +"	 插入队列" + data +"成功");
            } else {
                System.out.println(Thread.currentThread().getName() +"	 插入队列" + data +"失败");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName()+"	 大老板叫停了，表示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()+"	超过2秒钟没有取到蛋糕，消费退出");
                System.out.println();
                System.out.println();
                return ;
            }
            System.out.println(Thread.currentThread().getName()+"	消费队列"+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()+"	生产线程启动");
            try {
                myResource.myProduct();
            } catch (Exception e) {
                e.printStackTrace();
            }

        },"product").start();

        new Thread(() ->{
            System.out.println(Thread.currentThread().getName()+"	消费线程启动");
            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();
    }
}

运行结果见下图：