Java生产消费者模型——代码解析

我们将生产者、消费者、库存、和调用线程的主函数分别写进四个类中，通过抢夺非线程安全的数据集合来直观的表达在进行生产消费者模型的过程中可能出现的问题与解决办法。

我们假设有一个生产者，两个消费者来共同抢夺库存里的资源，而生产者和消费者都以线程来实现。

库存对象只有是唯一的才会出现抢夺一个资源的可能，所以为了使库存对象是唯一的，我们可以使用两种方法实现，单例模式和通过生产者和消费者的构造函数参数来初始化。

本次举例使用的是构造函数的方法，但代码中也注释出了单例模式的写法与使用。

先创建一个简单的生产消费者模型，查看它的运行结果。

• 库存类：

package producterac;

import java.util.arraylist;

public class warehouse {

    //存放非线程安全的数组的集合
    private arraylist<string> list = new arraylist<string>();
    
    
    /*
     * //创建单例模式使生产消费者操作的是同一库存对象 
     * private warehouse() {} 
     * //建立静态对象以在初始化的时候建立仅一个库存对象
     * private static warehouse wh = new warehouse();
     * 
     * //将方法设置为静态是因为在无法new库存对象的情况下， 
     * //我们可以通过将方法设定为静态来直接通过类名调用静态方法 
     * public static warehouse getinstance() {
     *  return wh;
     * }
     */
    
    
    //写生产者操作仓库的方法
    public void add() {
        if(list.size() < 20) {
            list.add("一个数据");
        }else {
            //数据存够之后直接返回,不运行存储数据的操作
            return;
        }
    }
    
    //写消费者操作仓库的操作
    public void get() {
        //判断集合中是否还有数据可以取出
        //如果不判断会造成集合越界
        if(list.size() > 0) {
            list.remove(0);
        }else {
            return;
        }
    }
    
}

• 生产者类：

package producterac;

public class producter extends thread{

    private string pname;
    
    //我们要使生产者和消费者操控同一个库存对象
    //也可以使用单例模式来建立库存对象
    private warehouse wh;
    public producter(string pname,warehouse wh) {
        this.pname = pname;
        this.wh = wh;
    }
    
    
    //重写run方法
    public void run() {
        while(true) {
            wh.add();
            system.out.println("生产者"+pname+"添加了一个货物");
            try {
                //使线程等待一会儿
                thread.sleep(200);
            } catch (interruptedexception e) {
                // todo auto-generated catch block
                e.printstacktrace();
            }
        }
    }
}

• 消费者类：

package producterac;

public class consumer extends thread{

    private string cname;
    
    //获取库存对象
    /* private warehouse wh = warehouse.getinstance(); */
    
    //我们要使生产者和消费者操控同一个库存对象
    //也可以使用单例模式来建立库存对象
    private warehouse wh;
    public consumer(string cname,warehouse wh) {
        this.cname = cname;
        this.wh = wh;
    }
    
    //重写run方法
    public void run() {
        while(true) {
            wh.get();
            system.out.println("消费者"+cname+"拿走了一个货物");
            try {
                //使线程等待一会儿
                thread.sleep(200);
            } catch (interruptedexception e) {
                // todo auto-generated catch block
                e.printstacktrace();
            }
        }
    }
}

• 主函数类：

package producterac;

public class main {

    public static void main(string[] args) {
        
        warehouse wh = new warehouse();
        producter p1 = new producter("1", wh);
        consumer c1 = new consumer("1", wh);
        consumer c2 = new consumer("2", wh);
        
        p1.start();
        c1.start();
        c2.start();
    }
}

部分运行结果：

 

 

我们看到出现 java.lang.arrayindexoutofboundsexception异常，说明消费者在拿走货物的时候集合越界没有拿到，所以出现了异常。

即使我们在库存的get()方法中判断了集合是否为空，但也还是出现了异常。原因是因为在两个线程同时访问一个对象的时候，有可能当线程1刚判断完集合不为空进入了if循环但还没有拿走货物的情况下，线程2也进行了get()方法先线程1一步拿走了最后的一个货物，然后当线程1想拿走货物的时候集合里已经没有了，这种情况下就会发生上述异常。

这就造成了线程抢夺资源时非安全的问题，那么我们可以将库存对象使用线程锁synchronized锁起来，这样在一个消费者访问库存对象的时候其他消费者无法访问库存对象，从而解决集合越界问题，使线程安全。

• 修改过的库存类（加入了synchronized修饰符的add()和get()方法）：

    //写生产者操作仓库的方法
    public synchronized void add() {
        if(list.size() < 20) {
            list.add("一个数据");
        }else {
            //数据存够之后直接返回,不运行存储数据的操作
            return;
        }
    }
    
    //写消费者操作仓库的操作
    public synchronized void get() {
        //判断集合中是否还有数据可以取出
        //如果不判断会造成集合越界
        if(list.size() > 0) {
            list.remove(0);
        }else {
            return;
        }
    }

 

使用synchronized修饰符修饰库存方法之后就不会报错了！

 

我们也可以将return替换为wait()方法让线程等待，将编写的生产消费者模型中的return修改为wait()。

• 修改过的库存类：　

    //写生产者操作仓库的方法
    public synchronized void add() {
        if(list.size() < 20) {
            list.add("一个数据");
        }else {
            try {
                //这个this指的是访问库存对象的线程wait，不是库存对象wait
                this.wait();
            } catch (interruptedexception e) {
                // todo auto-generated catch block
                e.printstacktrace();
            }
        }
    }
    
    //写消费者操作仓库的操作
    public synchronized void get() {
        //判断集合中是否还有数据可以取出
        //如果不判断会造成集合越界
        if(list.size() > 0) {
            list.remove(0);
        }else {
            try {
                //这个this指的是访问库存对象的线程wait，不是库存对象wait
                this.wait();
            } catch (interruptedexception e) {
                // todo auto-generated catch block
                e.printstacktrace();
            }
        }
    }

运行结果：

 

 

 我们会发现到最后所有的线程都会处于wait等待状态，运行到最后没有线程在执行了。所以我们需要在其中一个线程等待的时候将其他线程继续唤醒，保持系统的运行。

唤醒线程可以使用notify/notifyall()方法。

• 再次修改后的库存类：

    //写生产者操作仓库的方法
    public synchronized void add() {
        if(list.size() < 20) {
            list.add("一个数据");
        }else {
            try {
                //因为我们无法知道哪个线程是消费者线程，所以我们要将线程全部唤醒
                this.notifyall();
                //这个this指的是访问库存对象的线程wait，不是库存对象wait
                this.wait();
            } catch (interruptedexception e) {
                // todo auto-generated catch block
                e.printstacktrace();
            }
        }
    }
    
    //写消费者操作仓库的操作
    public synchronized void get() {
        //判断集合中是否还有数据可以取出
        //如果不判断会造成集合越界
        if(list.size() > 0) {
            list.remove(0);
        }else {
            try {
                //因为我们无法知道哪个线程是生产者线程，所以我们要将线程全部唤醒
                this.notifyall();
                //这个this指的是访问库存对象的线程wait，不是库存对象wait
                this.wait();
            } catch (interruptedexception e) {
                // todo auto-generated catch block
                e.printstacktrace();
            }
        }
    }

运行成功！说明我们这时候真正地实现了简单的生产消费者模型。

 

附：如果将完成的生产消费者模型中add()和get()方法的synchronized修饰符去掉，会发生如下错误。

 

 

将synchronized修饰符去掉后，发生了java.lang.illegalmonitorstateexception异常，原因是当线程1进入else要执行wait()方法的那个时刻，线程2也进入了库存对象中，致使当wait()方法真正执行的时候wait的是线程2而不是线程1，发生这种情况的时候就会发生上述异常。