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

深藏阁楼爱情的钟 2022-01-26 10:45 209阅读 0赞

# Java多线程之线程通信之生产者消费者传统版和阻塞队列版 #

--------------------

## 目录 ##

1.  线程通信之生产者消费者传统版
2.  线程通信之生产者消费者阻塞队列版

--------------------

### 1. 线程通信之生产者消费者传统版 ###

--------------------

题目： 一个初始值为零的变量，两个线程对其交替操作，一个加1一个减1，来五轮  
1.代码实现

import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    class ShareData {
          //资源类
        private int number = 0;
        private Lock lock = new ReentrantLock();
        private Condition condition = lock.newCondition();
    
        public void increment() throws Exception {
        
            lock.lock();
            try {
        
                //1.判断
                while (number != 0) {
        
                    //等待，不能生产
                    condition.await();
                }
                //2.干活
                number++;
                System.out.println(Thread.currentThread().getName() + "\t" + number);
                //通知唤醒
                condition.signalAll();
    
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                lock.unlock();
            }
        }
    
        public void decrement() throws Exception {
        
            lock.lock();
            try {
        
                //1.判断
                while (number == 0) {
        
                    //等待，不能生产
                    condition.await();
                }
                //2.干活
                number--;
                System.out.println(Thread.currentThread().getName() + "\t" + number);
                //通知唤醒
                condition.signalAll();
    
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                lock.unlock();
            }
        }
    
    
    }
    
    /**
     * 题目：
     * 一个初始值为零的变量，两个线程对其交替操作，一个加1一个减1，来五轮
     * <p>
     * 1.      线程  操作（方法）  资源类
     * 2.      判断  干活  通知
     * 3.      防止虚假唤醒
     */
    public class ProdConsumer_TraditionDemo {
        
        public static void main(String[] args) {
        
            ShareData shareData = new ShareData();
    
            new Thread(() -> {
        
                for (int i = 1; i <=5 ; i++) {
        
                    try {
        
                        shareData.increment();
                    } catch (Exception e) {
        
                        e.printStackTrace();
                    }
                }
            },"AA").start();
    
            new Thread(() -> {
        
                for (int i = 1; i <=5 ; i++) {
        
                    try {
        
                        shareData.decrement();
                    } catch (Exception e) {
        
                        e.printStackTrace();
                    }
                }
            },"BB").start();
        }
    }

2.编译结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70]

--------------------

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

题目： 一个初始值为零的变量，两个线程对其交替操作，一个加1一个减1，来五轮  
1.代码实现

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    class MyResource {
        
        private volatile boolean FLAG = true; //默认开启，进行生产+消费
        private AtomicInteger atomicInteger = new AtomicInteger();
    
        BlockingQueue<String> blockingQueue = null;
    
        public MyResource(BlockingQueue<String> blockingQueue) {
        
            this.blockingQueue = blockingQueue;
            System.out.println(blockingQueue.getClass().getName());
        }
    
        public void myProdu() 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 + "失败");
                }
                try {
        
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
            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()throws Exception{
        
            this.FLAG = false;
        }
    }
    
    /**
     *
     */
    public class ProdConsumer_BlockQueueDemo {
        
        public static void main(String[] args) throws Exception {
        
            MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
    
            new Thread(() -> {
        
                System.out.println(Thread.currentThread().getName() + "\t 生产线程启动");
                try {
        
                    myResource.myProdu();
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
            }, "Prod").start();
    
            new Thread(() -> {
        
                System.out.println(Thread.currentThread().getName() + "\t 消费线程启动");
                System.out.println();
                System.out.println();
                try {
        
                    myResource.myConsumer();
                    System.out.println();
                    System.out.println();
                } 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();
        }
    }

1.  执行结果  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70]: /images/20220126/00205e8132ff447da05df2f25f27cc39.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70 1]: /images/20220126/4d7f50dc6755425cb2b415699a103df2.png