java多线程：使用BlockingQueue（阻塞队列）实现生产者与消费者应用

深藏阁楼爱情的钟 2022-03-20 09:44 152阅读 0赞

### 前提概要 ###

#### 什么是BlokingQueue ####

BlockingQueue 是一个先进先出的队列（Queue），为什么说是阻塞（Blocking）的呢？是因为 BlockingQueue 支持当获取队列元素但是队列为空时，会阻塞等待队列中有元素再返回；也支持添加元素时，如果队列已满，那么等到队列可以放入新元素时再放入。

BlockingQueue 对插入操作、移除操作、获取元素操作提供了四种不同的方法用于不同的场景中使用：1、抛出异常；2、返回特殊值（null 或 true/false，取决于具体的操作）；3、阻塞等待此操作，直到这个操作成功；4、阻塞等待此操作，直到成功或者超时指定时间。

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

### 实现代码 ###

package com.oumuv.producerandconsumer;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingDeque;
    
    /** * 多线程生产者与消费者问题 * 生产者生产，消费者立刻消费 * 使用阻塞队列 {@link BlockingQueue}保存产品 */
    public class ProducerAndConsumer1 { 
    
    
        public static void main(String[] args) { 
            ProducerAndConsumer1 pac = new ProducerAndConsumer1();
            Storage storage = pac.new Storage();
            Producer p1 = pac.new Producer("生产者1", "小米手机", storage);
            Producer p2 = pac.new Producer("生产者2", "华为手机", storage);
            Consumer c1 = pac.new Consumer(storage);
    
            ExecutorService executorService = Executors.newCachedThreadPool();
            executorService.submit(p1);
            executorService.submit(p2);
            executorService.submit(c1);
        }
    
        /** * 生产者 */
        class Producer implements Runnable { 
            private String name;//生产者名称
            private String pname;//生产的产品
            private Storage storage;
    
            public Producer(String name, String pname, Storage storage) { 
                this.name = name;
                this.pname = pname;
                this.storage = storage;
            }
    
            @Override
            public void run() { 
                try { 
                    for (int i = 0; i < 15; i++) { 
                        Product p = new Product(pname, name);
                        storage.push(p);
                        Thread.sleep(100);
                    }
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        /** * 消费者 */
        class Consumer implements Runnable { 
            private Storage storage;
    
            public Consumer(Storage storage) { 
                this.storage = storage;
            }
    
            @Override
            public void run() { 
                try { 
                    while (true) { 
                        storage.pop();
                        Thread.sleep(500);
                    }
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        /** * 产品信息 */
        class Product { 
            private String name;
            private String from;//来自哪个生产者
            private int id = 0;
    
            public Product(String name, String from, int id) { 
                this.name = name;
                this.from = from;
                this.id = id;
            }
    
            public Product(String name, String from) { 
                this.name = name;
                this.from = from;
            }
    
            public String getName() { 
                return name;
            }
    
            public String getFrom() { 
                return from;
            }
    
            public int getId() { 
                return id;
            }
    
            public void setId(int id) { 
                this.id = id;
            }
        }
    
        /** * 产品仓库 */
        class Storage { 
    
            //阻塞队列
            private BlockingQueue<Product> storage = new LinkedBlockingDeque<>();
    
            private int index = 1;//用于生成产品编号
    
            /** * 消费 */
            protected void pop() throws InterruptedException { 
                Product product = storage.take();
                System.out.println(product.getFrom() + "的" + product.getName() + "--> id:" + product.getId() + "被消费了");
            }
    
            /** * 生产 */
            protected void push(Product product) throws InterruptedException { 
                storage.put(product);
                product.setId(index++);
                System.out.println(product.getFrom() + ",生产了" + product.getName() + "--> id:" + product.getId());
            }
        }
    }

### 可能输出的结果 ###

生产者1,生产了小米手机--> id:1
    生产者2,生产了华为手机--> id:2
    生产者1的小米手机--> id:1被消费了
    生产者1,生产了小米手机--> id:4
    生产者2,生产了华为手机--> id:3
    生产者2,生产了华为手机--> id:5
    生产者1,生产了小米手机--> id:6
    生产者1,生产了小米手机--> id:7
    生产者2,生产了华为手机--> id:8
    生产者2,生产了华为手机--> id:10
    生产者1,生产了小米手机--> id:9
    生产者2的华为手机--> id:2被消费了
    生产者1,生产了小米手机--> id:11
    生产者2,生产了华为手机--> id:12
    生产者2,生产了华为手机--> id:14
    生产者1,生产了小米手机--> id:13
    生产者2,生产了华为手机--> id:16
    生产者1,生产了小米手机--> id:15
    生产者2,生产了华为手机--> id:17
    生产者1,生产了小米手机--> id:18
    生产者1,生产了小米手机--> id:19
    生产者2,生产了华为手机--> id:20
    生产者2的华为手机--> id:3被消费了
    生产者2,生产了华为手机--> id:21
    生产者1,生产了小米手机--> id:22
    生产者2,生产了华为手机--> id:23
    生产者1,生产了小米手机--> id:24
    生产者2,生产了华为手机--> id:25
    生产者1,生产了小米手机--> id:26
    生产者2,生产了华为手机--> id:27
    生产者1,生产了小米手机--> id:28
    生产者2,生产了华为手机--> id:29
    生产者1,生产了小米手机--> id:30
    生产者1的小米手机--> id:4被消费了
    生产者2的华为手机--> id:5被消费了
    生产者1的小米手机--> id:6被消费了
    生产者1的小米手机--> id:7被消费了
    生产者2的华为手机--> id:8被消费了
    生产者1的小米手机--> id:9被消费了
    生产者2的华为手机--> id:10被消费了
    生产者1的小米手机--> id:11被消费了
    生产者2的华为手机--> id:12被消费了
    生产者1的小米手机--> id:13被消费了
    生产者2的华为手机--> id:14被消费了
    生产者1的小米手机--> id:15被消费了
    生产者2的华为手机--> id:16被消费了
    生产者2的华为手机--> id:17被消费了
    生产者1的小米手机--> id:18被消费了
    生产者1的小米手机--> id:19被消费了
    生产者2的华为手机--> id:20被消费了
    生产者2的华为手机--> id:21被消费了
    生产者1的小米手机--> id:22被消费了
    生产者2的华为手机--> id:23被消费了
    生产者1的小米手机--> id:24被消费了
    生产者2的华为手机--> id:25被消费了
    生产者1的小米手机--> id:26被消费了
    生产者2的华为手机--> id:27被消费了
    生产者1的小米手机--> id:28被消费了
    生产者2的华为手机--> id:29被消费了
    生产者1的小米手机--> id:30被消费了