JUC并发包——BlockingQueue 阻塞队列 解决生产者消费者问题

package com.bjsxt.commu2;

import com.bjsxt.commu1.ProductFactory2;

public class Test3 {

    public static void main(String[] args) {
        //创建并启动多个生产者和消费者线程
        final ProductFactory3 factory = new ProductFactory3(30);

        for (int i = 0; i < 20; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    int j=1;
                    while (true){
                        factory.produce("商品"+j);
                        j++;
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }, "生产者" + i).start();
        }

        for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true){
                        //消费一个商品
                        factory.consume();
                        try {
                            Thread.sleep(500);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }, "消费者" + i).start();
        }
    }
}

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.LinkedBlockingDeque;

  

  /**

 * 商品工厂

 */

  public class ProductFactory3 {

  //    private List<String> list = new ArrayList<String>();

//    private Lock lock = new ReentrantLock();

//    private Condition produceCond = lock.newCondition();

//    private Condition consumeCond = lock.newCondition();

  

    private BlockingQueue blockingQueue;

  

    private int max = 10;

  

    public ProductFactory3() {

  //        blockingQueue = new ArrayBlockingQueue(max);//固定长度的，一但指定不可改变

        blockingQueue = new LinkedBlockingDeque(max);//不指定上限就是无限长的

    }

  

    public ProductFactory3(int max) {

        this.max = max;

  //        blockingQueue = new ArrayBlockingQueue(max);//固定长度

        blockingQueue = new LinkedBlockingDeque(max);

    }

  

    /**

     * 生产商品

     * @param productName

     */

    public  void produce(String productName){

  //        lock.lock();

//        try {

//            //仓库满，就不在生产

//            while (list.size()==max){

//                System.out.println(Thread.currentThread().getName()+"仓库已满,目前商品数量"+"");

//                produceCond.await();

//            }

//            //生产商品

//            list.add(productName);

        //仓库满，提示仓库满了

            if (blockingQueue.size()==max){

                  System.out.println(Thread.currentThread().getName()+"仓库已满,目前商品数量"+blockingQueue.size());

            }

  

  //        blockingQueue.add(productName);//仓库满了就报错并中断该线程

        try {

            blockingQueue.put(productName);//仓库满了不报错，让该线程处于等待状态，并唤醒其他线程

            System.out.println(Thread.currentThread().getName()+"生产商品:" + productName+",目前商品数量"+blockingQueue.size());

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        //唤醒消费者，通知消费者消费

            /*没有指定唤醒哪个线程，消费者和生产者处在同一个等待队列，

                可能唤醒一个生产者,我们让生产者再判断，生产者发现仓库满的，继续沉睡，

                再唤醒别的生产者，我们让别的生产者再判断，别的生产者再判断发现仓库满的，

                别的生产者继续沉睡，直到唤醒一个消费者。

                如何让他再回去判断？把判断做成一个循环，不断反复判断，

                不符合就跳出循环，符合就执行循环继续沉睡。

             */

//            consumeCond.signal();

//        } catch (InterruptedException e) {

//            e.printStackTrace();

//        }finally {

//            lock.unlock();

//

//        }

    }

  

    /**

     * 消费商品

     */

    public  void consume() {

  //        lock.lock();

//        try {

//            while (list.size() == 0) {

//                System.out.println(Thread.currentThread().getName() + "仓库已经没有库存了，目前商品的数量：" +"");

//                consumeCond.await();

//            }

//            String productName = list.remove(0);

            if (blockingQueue.size() == 0) {

                System.out.println(Thread.currentThread().getName() + "仓库已经没有库存了，目前商品的数量：" +blockingQueue.size());

  

            }

  //            String productName = (String) blockingQueue.remove();//仓库没有库存了就报错，并中断线程

        try {

            String productName = (String) blockingQueue.take();//仓库没有库存了不报错，让该线程处于等待状态，并唤醒其他线程

            System.out.println(Thread.currentThread().getName() + "消费商品：" + productName + "，目前商品的数量：" +blockingQueue.size());

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

  //            //唤醒生产者，通知生产者生产

//            produceCond.signal();

//            //唤醒所有的生产者，但是只有一个生产者首先获取Lock,然后进入就绪队列，通过调度获得CPU

//            produceCond.signalAll();

//        } catch (InterruptedException e) {

//            e.printStackTrace();

//        }finally {

//            lock.unlock();

//

//        }

    }

}

抛出异常

特殊值

阻塞

超时

插入

add(e)

offer(e)

put(e)

offer(e, time, unit)

移除

remove()

poll()

take()

poll(time, unit)

检查

element()

peek()

不可用

不可用