java生产者消费者问题

时间:2022-05-23 yuuunfryan 人气:0

应用场景

假设仓库中只能存放一件产品，生产者将生产出来的产品放入仓库，消费者将仓库中产品取走消费

如果仓库中没有产品，则生产者将产品放入仓库，否则停止生产并等待，直到仓库中的产品被消费者取走为止

如果仓库中放有产品，则消费者可以将产品取走消费，否则停止消费并等待，直到仓库中再次放入产品为止

分析

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

对于生产者，没有生产产品之前，要通知消费者等待，而生产了产品之后，又需要马上通知消费者消费

对于消费者，在消费之后，要通知生产者已经结束消费，需要生产新的产品以供消费

在生产者消费者问题中，仅有synchronized是不够的

synchronized可阻止并发更新同一个共享资源，实现了同步

synchronized不能用来实现不同线程之间的消息传递（通信）

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

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

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

解决方法

管程法

生产者：负责生产数据的模块（可能是方法、对象、线程、进程）

消费者：负责处理数据的模块（可能是方法、对象、线程、进程）

缓冲区：消费者不能直接使用生产者的数据，他们之间有个“缓冲区”

生产者将生产好的数据放入缓冲区，消费者从缓冲区拿出数据

代码案例

//使用缓冲区解决生产者消费者模型
public class TestPC {
    public static void main(String[] args) {
        SynContainer container = new SynContainer();
        new Producer(container).start();
        new Consumer(container).start();
    }
}

class Producer extends Thread{
    SynContainer synContainer;
    public Producer(SynContainer synContainer){
        this.synContainer = synContainer;
    }
    //  生产
    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            synContainer.push(new Chicken(i));
            System.out.println("生产了"+i+"只鸡");
        }
    }
}

class Consumer extends Thread{
    SynContainer synContainer;
    public Consumer(SynContainer synContainer){
        this.synContainer = synContainer;
    }
    //  消费
    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            System.out.println("消费了-->"+synContainer.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;
    }
}

信号灯法

通过标志位 true 或者 false 来进行判断

代码案例

//信号灯法测试生产者消费者模型 即标志位
public class TestPC2 {
    public static void main(String[] args) {
        TV tv = new TV();
        new Player(tv).start();
        new Audiance(tv).start();
    }

}

//生产者-->演员
class Player extends Thread {
    TV tv;
    public Player(TV tv){
        this.tv = tv;
    }

    @Override
    public void run() {
        for (int i = 0; i < 20; i++) {
            if (i%2==0){
                this.tv.play("天天向上");
            }else {
                this.tv.play("抖音广告");
            }
        }
    }
}

//消费者-->观众
class  Audiance extends Thread{
    TV tv;
    public Audiance(TV tv){
        this.tv = tv;
    }

    @Override
    public void run() {
        for (int i = 0; i < 20; i++) {
            tv.watch();
        }
    }
}

//产品-->节目
class TV{
    //演员表演 观众等待 T
    //观众观看 演员等待 F
    String show;//表演的节目
    boolean flag = true;

    //表演
    public synchronized void play(String show){
        if (!flag){
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("演员表演了："+show);
        //通知观众
        this.notifyAll();
        this.show = show;
        this.flag = !this.flag;
    }

    //观看
    public synchronized void watch(){
        if (flag){
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("观看了"+show);
        //通知演员表演
        this.notifyAll();
        this.flag = !this.flag;
    }
}

总结

加载全部内容