生产者消费者问题

生产者/消费者 问题

生产者 / 消费者 问题 是线程模型中的经典问题

问题概述（为什么？）：

​ 生产者 和 消费者 在同一时间段内 共用 同一个存储空间 如果不加以协调 可能出现以下情况 存储空间已满 而生产者占用着它 消费者等待着 生产者 让出资源 从而去除产品 而 生产者 等待着 消费者 去消费产品 从而向空间中 新增产品 相互等待 产生死锁

产什么死锁的四个必要条件：

• 互斥条件 ： 一个资源一次只能被一个线程使用
• 请求与保持条件： 一个线程因请求资源 而 阻塞时 对已经获得的资源 保持 不放
• 不剥夺条件： 线程已经获得资源 在未使用完后之前 不能强行剥夺
• 循环等待条件：若干线程之间 形成一种 头尾相接的 循环等待资源关系
• 上述问题简而言之：同一时刻 统一资源 被两个线程(多个线程)使用 并且 两个线程相互等待对方执行结束 从而释放资源 另一方 接着使用 共享资源

将 共享资源 放入 共享缓冲区 中

生产者 —生产—> 缓冲区 — 消费 —> 消费者

为什么？

解耦

• 消费者 不直接 使用 生产者 如果某天代码需要生产者中的代码发生变化 不会影响到 消费者 代码的使用
• 将 生产者 消费者 两者 共同依赖 放入 共享缓存区中 两者不直接依赖 降低耦合度

支持并发

• 可以有多组 生产者 消费者 来缓冲区 获取资源

提高性能

• 生产者 各自只需要处理自身业务 不需要等待 消费者 前来获取已经生产好的

面试题 ： 线程通信问题

• 两个线程交替打印

class ShareDataOne {
    // 共享资源类
    // 定义一个变量
    private int number = 0;

    // 使用lock锁 代替 synchronized
    private final Lock lock = new ReentrantLock();

    // wait() - await();
    private Condition condition = lock.newCondition();

    // 生产者线程
    public synchronized void increment() throws InterruptedException {
    //public void increment() throws InterruptedException {
        // 判断
        //lock.lock(); 
      //  try {
            if (number != 0) {
            //while (number != 0) {
                this.wait();
      //          condition.await();
            }
            // 干活
            ++number;
            System.out.println(Thread.currentThread().getName() +"\t" + number);
            // 通知/唤醒
            this.notifyAll();
      //      condition.signalAll();
      //  } catch (InterruptedException e) {
      //      e.printStackTrace();
      //  } finally {
          //  lock.unlock();
        }
    }

    // 消费者线程
    public synchronized void decrement() throws InterruptedException {
        // 判断
       // lock.lock();
       // try {
            if (number == 0) {
            //while (number == 0) {
                this.wait();
        //        condition.await();
            }
            // 干活
            --number;
            System.out.println(Thread.currentThread().getName() + "\t" + number);
            // 通知
            this.notifyAll();
       //     condition.signalAll();
      //  } catch (InterruptedException e) {
      //      e.printStackTrace();
       // } finally {
       //     lock.unlock();
        }
    }

}

public class NotifyWaitDemo {

    public static void main(String[] args) {
        // 线程资源操作类
        ShareDataOne shareDataOne = new ShareDataOne();

        // 创建多个线程
        // 线程一：+
        new Thread(()->{
            // 生产
            for (int i = 0; i < 10; i++) {
                try {
                    shareDataOne.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"AA+").start();

        // 线程二：-
        new Thread(()->{
            // 生产
            for (int i = 0; i < 10; i++) {
                try {
                    shareDataOne.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"BB-").start();
    }
}

注意 上面代码中 生产者 消费者 是使用if判断的 当 有多组 生产者消费者存在的 情况下 将会出现 虚假唤醒 的 情况

public class NotifyWaitDemo {

    public static void main(String[] args) {
        // 线程资源操作类
        ShareDataOne shareDataOne = new ShareDataOne();

        // 创建多个线程
        // 线程一：+
        new Thread(()->{
            // 生产
            for (int i = 0; i < 10; i++) {
                try {
                    shareDataOne.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"AA+").start();

        // 线程二：-
        new Thread(()->{
            // 生产
            for (int i = 0; i < 10; i++) {
                try {
                    shareDataOne.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"BB-").start();

        // 当有多套生产者 消费者时 会存在 "虚假唤醒" 的 情况

        // 线程三：+
        new Thread(()->{
            // 生产
            for (int i = 0; i < 10; i++) {
                try {
                    shareDataOne.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"CC+").start();

        // 线程四：-
        new Thread(()->{
            // 生产
            for (int i = 0; i < 10; i++) {
                try {
                    shareDataOne.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"DD-").start();
    }
}

怎么办？

• 在1.5以前 将判断语句 从 if 改成 while 即可
• 在1.5以后 可以使用 Condition
  • Condition 中的 方法 对标 Synchronized
  • await() —- wait()
  • signal() —- notify()
  • signalAll() —- notifyAll()

赏

谢谢你请我吃糖果