生产者消费者模型

作者: Casin | 来源:发表于2018-09-16 15:21 被阅读0次

生产者/消费者模式实现

等待/通知模式最经典的案例就是“生产者/消费者模式”模式。但此模式在使用上有几种“变形”，还有一些小的注意事项，但原理都是基于wait/notify的。

一生产者与一消费者：操作值

模拟项目创建一个类p_c_test

/**
 * 输出结果：
get的值是：1494750000110_26368251399553
set的值是：1494750000110_26368251410228
get的值是：1494750000110_26368251410228
set的值是：1494750000110_26368251420081
get的值是：1494750000110_26368251420081
set的值是：1494750000110_26368251431987
get的值是：1494750000110_26368251431987
set的值是：1494750000110_26368251442250
get的值是：1494750000110_26368251442250
set的值是：1494750000110_26368251452103
get的值是：1494750000110_26368251452103
set的值是：1494750000110_26368251462367
get的值是：1494750000110_26368251462367
set的值是：1494750000110_26368251474273
 *结果分析：
例子展示一个消费者一个生产者，进行数据的交互，在控制台中打印的get和set是交互运行的。
 * @author jiaxing
 *
 */
package entity;
/**
 * 类说明： 创建生产者类
 *
 * @author: Casin
 * @Create: 2017-05-15 16:49
 * @HOME: https://qincasin.github.io/
 */
public class P{
    private String lock;
    public P(String lock) {
        super();
        this.lock=lock;
    }
    public void setValue(){
        try {
            synchronized(lock){
                if(!ValueObject.value.equals("")){
                    lock.wait();  //等待
                }
                //nanoTime():返回最准确的可用系统计时器的当前值，以毫微秒（纳秒）为单位。（）不可用于计算两时间差，具体看源码注释
                String value=System.currentTimeMillis()+"_"+System.nanoTime();
                System.out.println("set的值是："+value);
                ValueObject.value=value;
                lock.notify(); //唤醒
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

package entity;
/**
 * 类说明： 消费者类
 *
 * @author: Casin
 * @Create: 2017-05-15 18:28
 * @HOME: https://qincasin.github.io/
 */

public class C{
    private String lock;
    public C(String lock) {
        super();
        this.lock = lock;
    }
    public void getValue(){
        try {
            synchronized(lock){
                if(ValueObject.value.equals("")){
                    lock.wait();//等待
                }
                System.out.println("get的值是："+ValueObject.value);
                ValueObject.value="";
                lock.notify();//唤醒
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
package extthread;
import entity.P;
/**
 * 类说明： 创建线程P
 *
 * @author: Casin
 * @Create: 2017-05-15 18:32
 * @HOME: https://qincasin.github.io/
 */

public class ThreadP extends Thread {
    private P p;
    public ThreadP(P p) {
        super();
        this.p=p;
    }

    @Override
    public void run() {
        while (true){
            p.setValue();
        }
    }
}

package extthread;

import entity.C;

/**
 * 类说明： 线程C
 *
 * @author: Casin
 * @Create: 2017-05-15 18:37
 * @HOME: https://qincasin.github.io/
 */

public class ThreadC extends  Thread{
    private C r;

    public ThreadC(C r) {
        this.r = r;
    }

    @Override
    public void run() {
        super.run();
        while (true){
            r.getValue();
        }
    }
}

package test;
import entity.C;
import entity.P;
import extthread.ThreadC;
import extthread.ThreadP;
/**
 * 类说明： 测试类
 *
 * @author: Casin
 * @Create: 2017-05-15 18:40
 * @HOME: https://qincasin.github.io/
 */

public class Run {
    public static void main(String[] args) {
        String lock = new String("");
        P p = new P(lock);
        C r = new C(lock);
        ThreadP threadP = new ThreadP(p);
        ThreadC threadC = new ThreadC(r);
        threadP.start();
        threadC.start();
    }
}

但如果在此例子的基础上，设计出多个生产者和多个消费者，那么在运行的过程中极有可能出现“假死”的情况，也就是所有的线程都呈WAITING等待状态。

多生产者多消费者：操作值-假死

“假死”的现象其实就是线程进入WAITING等待状态，如果全部线程都进入WAITING状态，则程序就不在执行任何业务功能了，整个项目呈停止状态。这在使用生产者与消费者模式时经常遇到。

模拟项目，创建类p_c_allWait
代码如下：

package entity;

/**
 * 类说明： 生产者
 *
 * @author: Casin
 * @Create: 2017-05-15 18:48
 * @HOME: https://qincasin.github.io/
 */

public class P {
    private String lock;

    public P(String lock) {
        this.lock = lock;
    }
    public void setValue(){
        try {
            synchronized (lock){
                if(!ValueObject.value.equals("")){
                    System.out.println("生产者"+Thread.currentThread().getName()+" WAITINFG了※ ");
                    lock.wait();
                }
                System.out.println("生产者"+Thread.currentThread().getName()+" RUNNABLE了 ");
                String value=System.currentTimeMillis()+"_"+System.nanoTime();
                ValueObject.value=value;
                lock.notify();
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

package entity;

/**
 * 类说明： 消费者
 *
 * @author: Casin
 * @Create: 2017-05-15 18:52
 * @HOME: https://qincasin.github.io/
 */

public class C {
    private String lock;

    public C(String lock) {
        this.lock = lock;
    }
    public void getValue(){
                try {
                    synchronized (lock){
                        while(ValueObject.value.equals("")){
                            System.out.println("消费者"+Thread.currentThread().getName()+" WAITING了☆");
                            lock.wait();
                        }
                        System.out.println("消费者"+Thread.currentThread().getName()+" RUNNABLE了");
                        ValueObject.value="";
                        lock.notify();
                    }
                }catch (InterruptedException e){
                    e.printStackTrace();
        }
    }
}
package entity;

/**
 * 类说明： 存储值的对象
 *
 * @author: Casin
 * @Create: 2017-05-15 18:48
 * @HOME: https://qincasin.github.io/
 */

public class ValueObject {
    public static String value="";
}
package extthread;

import entity.P;

/**
 * 类说明： 线程p
 *
 * @author: Casin
 * @Create: 2017-05-15 18:54
 * @HOME: https://qincasin.github.io/
 */

public class ThreadP extends Thread{
    private P p;

    public ThreadP(P p) {
        super();
        this.p = p;
    }


    @Override
    public void run() {
        super.run();
        while (true){
            p.setValue();
        }
    }
}
package extthread;

import entity.C;

/**
 * 类说明： ThreadC线程
 *
 * @author: Casin
 * @Create: 2017-05-15 18:55
 * @HOME: https://qincasin.github.io/
 */

public class ThreadC extends Thread{
    private C r;

    public ThreadC(C r) {
        super();
        this.r = r;
    }

    @Override
    public void run() {
        super.run();
        while (true){
            r.getValue();
        }
    }
}
package test;

import entity.C;
import entity.P;
import extthread.ThreadC;
import extthread.ThreadP;
/**
 *
 消费者消费者1 WAITING了☆
 消费者消费者2 WAITING了☆
 生产者生产者1 RUNNABLE了
 生产者生产者1 WAITINFG了※
 生产者生产者2 RUNNABLE了
 生产者生产者2 WAITINFG了※
 消费者消费者1 RUNNABLE了
 消费者消费者1 WAITING了☆
 消费者消费者2 WAITING了☆
 main - RUNNABLE
 Monitor Ctrl-Break - RUNNABLE
 生产者1 - WAITING
 消费者1 - WAITING
 生产者2 - WAITING
 消费者2 - WAITING
 *解释：输出结果中※符号代表本线程进入等待状态，需要额外注意这样的执行结果
 * 在代码中确实已经通过wait/notify进行通信了，但不保证notify唤醒的是异类，也许是同类，
 * 比如“生产者”唤醒“生产者”，或“消费者”唤醒“消费者”这样的情况。
 * 如果按这样的情况运行的比例积少成多，就会导致所有的线程都不能继续运行下去，大家都在等待，都呈WAITING状态，程序最后也就成“假死”状态，不能继续运行下去了。
 */

/**
 * 类说明： 测试类
 *
 * @author: Casin
 * @Create: 2017-05-15 18:56
 * @HOME: https://qincasin.github.io/
 */

public class Run {
    public static void main(String[] args) throws InterruptedException {
        String lock = new String("");
        P p = new P(lock);
        C r = new C(lock);
        ThreadP[]  threadP = new ThreadP[2];
        ThreadC[] threadC =new ThreadC[2];
        for (int i=0;i<2;i++){
            threadP[i] = new ThreadP(p);
            threadP[i].setName("生产者"+(i+1));
            threadC[i] = new ThreadC(r);
            threadC[i].setName("消费者"+(i+1));
            threadP[i].start();
            System.out.println("-------------11111--");
            threadC[i].start();
        }
        Thread.sleep(5000);
        //getThreadGroup():返回此线程所属的线程组，如果线程已经停止，此方法返回null
        //activeCount返回当前线程组以及子组中活动线程数的估计
        //enumerate():将当前线程的线程组及其子组中的每个活动线程复制到指定的数组中
        //getState():返回线程的当前状态--->总共六种状态
        Thread[] threadArray = new Thread[Thread.currentThread().getThreadGroup().activeCount()];
        Thread.currentThread().getThreadGroup().enumerate(threadArray);
        for(int i=0;i<threadArray.length;i++){
            System.out.println(threadArray[i].getName()+" - "+threadArray[i].getState());
        }
    }
}

多生产者多消费者：操作值

解决“假死”的方法，将上面的项目p_c_allWait中P.java和C.java文件中的notify()改成notifyAll()就可以解决了。原理是：不光同志同类线程，也包括异类。这样就至于出现假死的状态了，程序会一直运行下去。

一生产者一消费者：操作栈

本例子是使生产者想堆栈List对象中放入对象，使消费者从List堆栈中取出数据，List最大容量是1，实验环境只有一个生产者与一个消费者。

创建项目stack_1

package entity;

import java.util.ArrayList;
import java.util.List;

/**
 * 类说明：
 *
 * @author: Casin
 * @Create: 2017-05-15 20:26
 * @HOME: https://qincasin.github.io/
 */

public class MyStack {
    private List list = new ArrayList();
    //入栈
    synchronized public void push(){
        try {
            if(list.size()==1){
                this.wait();//当栈中元素为1时进行等待操作
            }
            list.add("anyString="+Math.random());
            this.notify();//唤醒
            System.out.println("push="+list.size());
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
    //出栈
    synchronized public String pop(){
        String returnValue="";
        try {
            if (list.size()==0){
                System.out.println("pop操作中的"+Thread.currentThread().getName()+" 线程呈wait状态");
                this.wait();
            }
            returnValue=""+list.get(0);
            list.remove(0);
            this.notify();
            System.out.println("pop="+list.size());
        }catch (InterruptedException e){
            e.printStackTrace();
        }
        return returnValue;
    }
}
package extthread;

import service.C;

/**
 * 类说明： 消费者线程用于pop操作
 *
 * @author: Casin
 * @Create: 2017-05-15 20:39
 * @HOME: https://qincasin.github.io/
 */

public class C_Thread extends Thread {
    private C c;

    public C_Thread(C c) {
        this.c = c;
    }

    @Override
    public void run() {
        super.run();
        while (true){
            c.popService();
        }
    }
}
package extthread;

import service.P;

/**
 * 类说明： 生产者线程push操作
 *
 * @author: Casin
 * @Create: 2017-05-15 20:34
 * @HOME: https://qincasin.github.io/
 */

public class P_Thread extends Thread {
    private P p;

    public P_Thread(P p) {
        this.p = p;
    }

    @Override
    public void run() {
        super.run();
        while (true){
            p.pushService();
        }
    }
}
package service;

import entity.MyStack;

/**
 * 类说明： 消费者用于pop操作
 *
 * @author: Casin
 * @Create: 2017-05-15 20:37
 * @HOME: https://qincasin.github.io/
 */

public class C {
    private MyStack myStack;

    public C(MyStack myStack) {
        this.myStack = myStack;
    }
    public void popService(){
        System.out.println("pop="+myStack.pop());
    }
}
package service;

import entity.MyStack;

/**
 * 类说明： 消费者用于pop操作
 *
 * @author: Casin
 * @Create: 2017-05-15 20:37
 * @HOME: https://qincasin.github.io/
 */

public class C {
    private MyStack myStack;

    public C(MyStack myStack) {
        this.myStack = myStack;
    }
    public void popService(){
        System.out.println("pop="+myStack.pop());
    }
}

package test;

import entity.MyStack;
import extthread.C_Thread;
import extthread.P_Thread;
import service.C;
import service.P;
/**
 * 输出结果：
 push=1
 pop=0
 pop=anyString=0.6610059962956492
 pop操作中的Thread-1 线程呈wait状态
 push=1
 pop=0
 pop=anyString=0.17516237577017224
 pop操作中的Thread-1 线程呈wait状态
 push=1
 pop=0
 pop=anyString=0.004120762121210819
 pop操作中的Thread-1 线程呈wait状态
 push=1
 pop=0
 pop=anyString=0.3785755506701072
 pop操作中的Thread-1 线程呈wait状态
 *解释：
 * 结果中显示消费者放push一个，则消费者pop一个，当没有push时，则消费者呈现waiting状态
 * 容器size()的值不会大于1，这也是例子想要实现的效果，值在0和1之间进行交替，也就是生产者和消费者这两个过程在交替执行。
 */

/**
 * 类说明： 测试一生产者一消费者：操作栈类
 *
 * @author: Casin
 * @Create: 2017-05-15 20:41
 * @HOME: https://qincasin.github.io/
 */

public class Run {
    public static void main(String[] args) {
        MyStack myStack = new MyStack();
        P p = new P(myStack);
        C c = new C(myStack);
        P_Thread p_thread = new P_Thread(p);
        C_Thread c_thread = new C_Thread(c);
        p_thread.start();
        c_thread.start();
    }
}

一生产者多消费者--操作栈：解决wait条件改变与假死

本例是使用一个生产者向堆栈List对象中放入数据，而多个消费者从List堆栈中取出数据。List最大容量仍然还是1

创建新的项目，将上个项目中的代码全部复制到新项目中；

代码如下：

解释：

当条件发生改变时并没有得到及时的相应，会导致多个呈wait状态的线程都被唤醒，继而执行list.remove(0)代码而出现异常。

解决办法：将Mystack中的if修改为while。

但是这时会出现程序假死状态

因此使用notifyAll()解决假死状态

package entity;

import java.util.ArrayList;
import java.util.List;

/**
 * 类说明：
 *
 * @author: Casin
 * @Create: 2017-05-15 20:26
 * @HOME: https://qincasin.github.io/
 */

public class MyStack {
    private List list = new ArrayList();
    //入栈
    synchronized public void push(){
        try {
            while(list.size()==1){
                this.wait();//当栈中元素为1时进行等待操作
            }
            list.add("anyString="+Math.random());
            this.notifyAll();//唤醒
            System.out.println("push="+list.size());
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
    //出栈
    synchronized public String pop(){
        String returnValue="";
        try {
            while (list.size()==0){
                System.out.println("pop操作中的"+Thread.currentThread().getName()+" 线程呈wait状态");
                this.wait();
            }
            returnValue=""+list.get(0);
            list.remove(0);
            this.notifyAll();
            System.out.println("pop="+list.size());
        }catch (InterruptedException e){
            e.printStackTrace();
        }
        return returnValue;
    }
}
package test;

import entity.MyStack;
import extthread.C_Thread;
import extthread.P_Thread;
import service.C;
import service.P;
/**
 * 输出结果：
 push=1
 pop=0
 pop=anyString=0.39424855100604783
 pop操作中的Thread-3 线程呈wait状态
 pop操作中的Thread-5 线程呈wait状态
 pop操作中的Thread-4 线程呈wait状态
 pop操作中的Thread-2 线程呈wait状态
 push=1
 pop=0
 *解释：
 * 程序会一直执行下去，当生产者push一个后，那么多个消费者只有一个会pot，其余的全部等待。
 * 解决了wait条件改变与假死的bug
 */

/**
 * 类说明： 测试一生产者多消费者：操作栈类
 *
 * @author: Casin
 * @Create: 2017-05-15 20:41
 * @HOME: https://qincasin.github.io/
 */

public class Run {
    public static void main(String[] args) {
        MyStack myStack = new MyStack();
        P p = new P(myStack);
        C c1 = new C(myStack);
        C c2 = new C(myStack);
        C c3 = new C(myStack);
        C c4 = new C(myStack);
        C c5 = new C(myStack);
        //启动一个生产者
        P_Thread p_thread = new P_Thread(p);
        p_thread.start();
        //启动多个消费者
        C_Thread c_thread1 = new C_Thread(c1);
        C_Thread c_thread2 = new C_Thread(c2);
        C_Thread c_thread3 = new C_Thread(c3);
        C_Thread c_thread4 = new C_Thread(c4);
        C_Thread c_thread5 = new C_Thread(c5);
        c_thread1.start();
        c_thread2.start();
        c_thread3.start();
        c_thread4.start();
        c_thread5.start();
    }
}

多生产者与一消费者：操作栈

本例子展示使用消费者向堆栈List对象中放入数据，使用消费者从List堆栈中取出数据。
List最大容量还是1

创建新项目，将上一个项目全部复制到新项目中，修改Run方法中代码

package test;

import entity.MyStack;
import extthread.C_Thread;
import extthread.P_Thread;
import service.C;
import service.P;

/**
 * 类说明： 多生产者与一消费者：操作栈
 *
 * @author: Casin
 * @Create: 2017-05-16 19:37
 * @HOME: https://qincasin.github.io/
 */

public class Run {
    public static void main(String[] args) {
       MyStack myStack = new MyStack();
        //创建多个生产者
         P p1 = new P(myStack);
        P p2 = new P(myStack);
        P p3 = new P(myStack);
        P p4 = new P(myStack);
        P p5 = new P(myStack);
        P p6 = new P(myStack);
        P_Thread p_thread1 = new P_Thread(p1);
        P_Thread p_thread2 = new P_Thread(p2);
        P_Thread p_thread3 = new P_Thread(p3);
        P_Thread p_thread4 = new P_Thread(p4);
        P_Thread p_thread5 = new P_Thread(p5);
        P_Thread p_thread6 = new P_Thread(p6);
        p_thread1.start();
        p_thread2.start();
        p_thread3.start();
        p_thread4.start();
        p_thread5.start();
        p_thread6.start();
        //创建一个消费者
        C c = new C(myStack);
        C_Thread c_thread = new C_Thread(c);
        c_thread.start();
    }
}

最后结果显示正常

多生产者与多消费者：操作栈

本例子是使用生产者向栈List对象中放入数据，使用消费者从List栈中取出数据。List最大容量是1，实验环境是多个生产者与多个生产者。
代码比较简单，只需要在Run类中多添加几个消费者就可以了。输出结果也是可想而知，当一个生产者push一个后，那么消费者只有一个可以pop，其余呈现waiting状态。

生产者消费者模型

生产者/消费者模式实现

一生产者与一消费者：操作值

多生产者多消费者：操作值-假死

多生产者多消费者：操作值

一生产者一消费者：操作栈

一生产者多消费者--操作栈：解决wait条件改变与假死

多生产者与一消费者：操作栈

多生产者与多消费者：操作栈

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读