java-生产者消费者问题以及解决办法

2023-05-16

文章目录

- 1.生产者消费者问题概述
- 2.生产者消费者问题的解决办法
- - 2.1 解决思路
  - 2.2 实现方法
  - 2.3 代码实现
  - - 2.3.1 wait()和nofity()方法
    - 2.3.2 await()/signal()方法
    - 2.3.3 BlockingQueue阻塞队列方法
    - 2.3.4 Semaphore信号量
    - 2.3.5 管道
    - - 2.4.5.1 PipedInoutStream/PipedOutputStream（操作字节流）
      - 2.4.5.2 PipedReader/PipedWriter（操作字符流）

1.生产者消费者问题概述

生产者消费者问题是多线程同步的一个经典问题。生产者消费者同时使用一块缓冲区，生产者生产商品放入缓冲区，消费者从缓冲区取出商品。我们需要保证的是，当缓冲区满时，生产者不可生产商品；当缓冲区为空时，消费者不可取出商品。

这是一个线程同步问题，生产者和消费者共享同一个资源，并且生产者和消费者之间相互依赖，互为条件。

对于生产者，没有生产产品之前，要通知消费者等待，而生产了产品之后，又需要马上通知消费者消费
对于消费者，在消费之后，要通知生产者已经结束消费，需要生产新的产品以供消费。
在生产者消费者问题中，仅有synchronized是不够的
- synchronized可阻止并发更新同一个共享资源，实现了同步
- synchronized不能用来实现不同线程之间的消息传递（通信）

Java提供了几个方法解决线程之间的通信问题

方法名	作用
wait()	表示线程一直等待，直到其他线程通知，与sleep不同，会释放锁
wait(long timeout)	指定等待的毫秒数
notify()	唤醒一个处于等待状态的线程
notifyAll()	唤醒同一个对象上所有调用wait()方法的线程，优先级别高的线程优先调度

注意：均是Object类的方法，都只能在同步方法或者同步代码块中使用，否则会抛出异常IIIegalMonitorStateException

2.生产者消费者问题的解决办法

2.1 解决思路

采用某种机制保护生产者消费者之间的同步，有较高的效率，并且易于实现，代码的可控制性较好，属于常用的方式
在生产者和消费者之间建立一个管道，管道缓冲区不易控制，被传输数据对象不易于封装等，实用性不强。

解决核心在于保证同一资源被多个线程并发访问时的完整性，常用的同步方法时采用信号或者加锁机制，保证资源在任意时刻至多被一个线程访问

2.2 实现方法

wait()和nofity()方法
await()和signal()方法
BlockingQueue阻塞队列方法
管道法
信号量

2.3 代码实现

2.3.1 wait()和nofity()方法

当缓冲区已满时，生产者线程停止执行，放弃锁，使自己处于等状态，让其他线程执行；
当缓冲区已空时，消费者线程停止执行，放弃锁，使自己处于等状态，让其他线程执行。

当生产者向缓冲区放入一个产品时，向其他等待的线程发出可执行的通知，同时放弃锁，使自己处于等待状态；
当消费者从缓冲区取出一个产品时，向其他等待的线程发出可执行的通知，同时放弃锁，使自己处于等待状态。
示例:

//测试生产者消费者模型--利用缓冲区解决
public class TestProductAndConsume1 {
    public static void main(String[] args) {
        SynContainer synContainer = new SynContainer();

        new Product(synContainer).start();
        new Consumer(synContainer).start();
    }
}

//生产者
class Product extends Thread {
    SynContainer container;

    public Product(SynContainer container){
        this.container = container;
    }

    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            container.push(new Chicken(i));
            System.out.println("生产了第" + i + "只鸡");
        }
    }
}

//消费者
class Consumer extends Thread {
    SynContainer container;

    public Consumer(SynContainer container){
        this.container = container;
    }

    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            System.out.println("消费了" + container.pop().id + "只鸡");
        }
    }
}

//产品
class Chicken {
    int id;

    public Chicken(int id) {
        this.id = id;
    }
}

//缓冲区
class SynContainer {
    //需要一个容器大小
    Chicken[] chickens = new Chicken[10];
    //容器计数器
    int count = 0;

    //生产者放入产品
    public synchronized void push(Chicken chicken){
        //容器满了，等待消费者消费
        if (count == chickens.length){
            //通知消费者消费，生产者等待
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //如果容器没有满，放入产品
        chickens[count] = chicken;
        count++;

        //通知消费者消费
        this.notifyAll();
    }

    //消费者消费产品
    public synchronized Chicken pop(){
        //判断能否消费
        if (count == 0){
            //等待生产者生产，消费者消费
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //如果可以消费
        count--;
        Chicken chicken = chickens[count];

        //吃完了，通知生产者生产
        this.notifyAll();
        return chicken;
    }
}

notify()方法可使所有正在等待队列中等待同一共享资源的"全部"线程从等待状态退出，进入可运行状态。此时，优先级最高的哪个线程最先执行，但也有可能是随机执行的，这要取决于JVM虚拟机的实现。即最终也只有一个线程能被运行，上述线程优先级都相同，每次运行的线程都不确定是哪个，后来给线程设置优先级后也跟预期不一样，主要看JVM具体实现。

2.3.2 await()/signal()方法

在JDK5中，用ReenteantLock和Condition可以实现等待/通知模型，具有更大的灵活性。通过在Lock对象上调用newCondition()方法，将条件变量和一个锁对象进行绑定，进而控制并发程序访问竞争资源的安全。

示例：（在这里只需要改动缓冲区即仓库类）

//缓冲区
class SynContainer {
    //需要一个容器大小
    Chicken[] chickens = new Chicken[10];
    //容器计数器
    int count = 0;

    // 锁
    private final Lock lock = new ReentrantLock();
    // 仓库满的条件变量
    private final Condition full = lock.newCondition();
    // 仓库空的条件变量
    private final Condition empty = lock.newCondition();

    //生产者放入产品
    public void push(Chicken chicken){
        //获得锁
        lock.lock();
        try {
            //容器满了，等待消费者消费
            if (count == chickens.length) {
                //通知消费者消费，生产者等待
                try {
                    full.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //如果容器没有满，放入产品
            chickens[count] = chicken;
            count++;

            //通知消费者消费
            empty.signalAll();
        } finally {
            lock.unlock();//释放锁
        }
    }

    //消费者消费产品
    public Chicken pop(){
        //获得锁
        lock.lock();
        Chicken chicken = null;
        try {
            //判断能否消费
            if (count == 0) {
                //等待生产者生产，消费者消费
                try {
                    empty.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //如果可以消费
            count--;
            chicken = chickens[count];

            //吃完了，通知生产者生产
            full.signalAll();
        } finally {
            lock.unlock();
        }
        return chicken;
    }
}

2.3.3 BlockingQueue阻塞队列方法

BlockingQueue是JDk5新增内容，它是一个已经在内部实现了同步的队列，实现方式采用的是我们第2种await()/signal()方法。它可以在生成对象时指定容量大小，用于阻塞操作的是put()和take()方法。

put()方法：类似于我们上面的生产者线程，容量达到最大时，自动阻塞。

take()方法：类似于我们上面的消费者线程，容量为0时，自动阻塞。

示例：（在这里只需要改动缓冲区即仓库类）

//缓冲区
class SynContainer {
    // 仓库存储的载体
    private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<>(10);

    //生产者放入产品
    public void push(Chicken chicken) {
        try {
            list.put(chicken);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //消费者消费产品
    public Chicken pop() {
        Chicken chicken = null;
        try {
            chicken = (Chicken) list.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return chicken;
    }
}

可能会出现put()或take()和System.out.println()输出不匹配的情况，是由于它们之间没有同步造成的。BlockingQueue可以放心使用，这可不是它的问题，只是在它和别的对象之间的同步有问题。

2.3.4 Semaphore信号量

Semaphonre是一种基于计数的信号量。它可以设定一个阈值，基于此，多个线程竞争获取许可对象，做完自己的申请后归还，超过阈值后，线程申请许可信号将会被阻塞。Semaphore可以用来构建一些对象池，比如数据库连接池，我们也可以创建计数为1的Semaphore,将其作为一种类似互斥锁的机制，这也叫二元信号量，表示两种互斥状态。计数为0的Semphore是可以release的，然后就可以acquire（即一开始使线程阻塞从而完成其他执行）。

示例：（在这里只需要改动缓冲区即仓库类）

//缓冲区
class SynContainer {
    // 仓库存储的载体
    private LinkedList<Object> list = new LinkedList<Object>();

    // 仓库的最大容量
    final Semaphore notFull = new Semaphore(10);
    // 将线程挂起，等待其他来触发
    final Semaphore notEmpty = new Semaphore(0);
    // 互斥锁
    final Semaphore mutex = new Semaphore(1);

    //生产者放入产品
    public void push(Chicken chicken) {
        try {
            notFull.acquire();
            mutex.acquire();
            list.add(chicken);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            mutex.release();
            notEmpty.release();
        }
    }

    //消费者消费产品
    public Chicken pop() {
        Chicken chicken = null;
        try {
            notEmpty.acquire();
            mutex.acquire();
            chicken = (Chicken) list.remove();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            mutex.release();
            notFull.release();
        }
        return chicken;
    }
}

2.3.5 管道

管道是一种特殊的流，用于不同线程间直接传送数据，一个线程发送数据到输出管道，另一个线程从输入管道读数据。

inputStream.connect(outputStream)或outputStream.connect(inputStream)作用是使两个Stream之间产生通信链接，这样才可以将数据进行输入和输出。

这种方式只适用于两个线程之间通信，不适合多个线程之间通信。

2.4.5.1 PipedInoutStream/PipedOutputStream（操作字节流）

示例：

//测试生产者消费者问题--管道-PipedInoutStream/PipedOutputStream（操作字节流）
public class TestProductAndConsume5 {
    public static void main(String[] args) {
        Producer producer = new Producer();
        Consumer consumer = new Consumer();
        Thread thread1 = new Thread(producer);
        Thread thread2 = new Thread(consumer);
        try {
            producer.getPipedOutputStream().connect(consumer.getPipedInputStream());
            thread2.start();
            thread1.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

class Producer implements Runnable {

    private PipedOutputStream pipedOutputStream;

    public Producer() {
        pipedOutputStream = new PipedOutputStream();
    }

    public PipedOutputStream getPipedOutputStream() {
        return pipedOutputStream;
    }

    @SneakyThrows
    @Override
    public void run() {
        try {
            for (int i = 0; i < 5; i++) {
                pipedOutputStream.write(("This is a test , ID = " + i + "!\n").getBytes());
            }
            pipedOutputStream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

class Consumer implements Runnable {

    private PipedInputStream pipedInputStream;

    public Consumer() {
        pipedInputStream = new PipedInputStream();
    }

    public PipedInputStream getPipedInputStream() {
        return pipedInputStream;
    }

    @SneakyThrows
    @Override
    public void run() {
        int length = -1;
        byte[] buffer = new byte[1024];
        try {
            while ((length = pipedInputStream.read(buffer)) != -1) {
                System.out.println(new String(buffer, 0, length));
            }
            pipedInputStream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

2.4.5.2 PipedReader/PipedWriter（操作字符流）

示例与操作字节流基本一致，只是替换相应流，这里不再贴代码

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)