首页 > 代码库 > 并发协作:多线程中的生产者与消费者模型

并发协作:多线程中的生产者与消费者模型

对于多线程程序来说,不管任何编程语言,生产者和消费者模型都是最经典的。就像学习每一门编程语言一样,Hello World!都是最经典的例子。

 

实际上,准确说应该是“生产者-消费者-仓储”模型,离开了仓储,生产者消费者模型就显得没有说服力了。

对于此模型,应该明确一下几点:

1、生产者仅仅在仓储未满时候生产,仓满则停止生产。

2、消费者仅仅在仓储有产品时候才能消费,仓空则等待。

3、当消费者发现仓储没产品可消费时候会通知生产者生产。

4、生产者在生产出可消费产品时候,应该通知等待的消费者去消费。

此模型将要结合java.lang.Object的wait与notify、notifyAll方法来实现以上的需求。这是非常重要的。

 

看下面的代码:<引:http://www.cnblogs.com/riskyer/p/3263032.html>

/** * Java线程:并发协作-生产者消费者模型 *  */public class Test {    public static void main(String[] args) {        Godown godown = new Godown(30);        Consumer c1 = new Consumer(50, godown);        Consumer c2 = new Consumer(20, godown);        Consumer c3 = new Consumer(30, godown);        Producer p1 = new Producer(10, godown);        Producer p2 = new Producer(10, godown);        Producer p3 = new Producer(10, godown);        Producer p4 = new Producer(10, godown);        Producer p5 = new Producer(10, godown);        Producer p6 = new Producer(10, godown);        Producer p7 = new Producer(80, godown);        c1.start();        c2.start();        c3.start();        p1.start();        p2.start();        p3.start();        p4.start();        p5.start();        p6.start();        p7.start();    }}/** * 仓库 */class Godown {    public static final int max_size = 100;// 最大库存量    public int curnum; // 当前库存量    Godown() {    }    Godown(int curnum) {        this.curnum = curnum;    }    /**     * 生产指定数量的产品     *      * @param neednum     */    public synchronized void produce(int neednum) {        // 测试是否需要生产        while (neednum + curnum > max_size) {            System.out.println("------要生产的产品数量" + neednum + "超过剩余库存量" + (max_size - curnum) + ",暂时不能执行生产任务!");            try {                // 当前的生产线程等待                wait();            } catch (InterruptedException e) {                e.printStackTrace();            }        }        // 满足生产条件,则进行生产,这里简单的更改当前库存量        curnum += neednum;        System.out.println("已经生产了" + neednum + "个产品,现仓储量为" + curnum);        // 唤醒在此对象监视器上等待的所有线程        notifyAll();    }    /**     * 消费指定数量的产品     *      * @param neednum     */    public synchronized void consume(int neednum) {        // 测试是否可消费        while (curnum < neednum) {            System.out.println("-----准备消费" + neednum + ",当前剩余" + curnum + ",暂不能消费!");            try {                // 当前的生产线程等待                wait();            } catch (InterruptedException e) {                e.printStackTrace();            }        }        // 满足消费条件,则进行消费,这里简单的更改当前库存量        curnum -= neednum;        System.out.println("已经消费了" + neednum + "个产品,现仓储量为" + curnum);        // 唤醒在此对象监视器上等待的所有线程        notifyAll();    }}/** * 生产者 */class Producer extends Thread {    private int neednum; // 生产产品的数量    private Godown godown; // 仓库    Producer(int neednum, Godown godown) {        this.neednum = neednum;        this.godown = godown;    }    public void run() {        // 生产指定数量的产品        godown.produce(neednum);    }}/** * 消费者 */class Consumer extends Thread {    private int neednum; // 生产产品的数量    private Godown godown; // 仓库    Consumer(int neednum, Godown godown) {        this.neednum = neednum;        this.godown = godown;    }    public void run() {        // 消费指定数量的产品        godown.consume(neednum);    }}

 

执行结果如下:(注意:每次运行的结果都不同)

情况1:死锁情况

-----准备消费50,当前剩余30,暂不能消费!
已经消费了30个产品,现仓储量为0
-----准备消费20,当前剩余0,暂不能消费!
已经生产了10个产品,现仓储量为10
-----准备消费50,当前剩余10,暂不能消费!
已经生产了10个产品,现仓储量为20
已经生产了10个产品,现仓储量为30
已经消费了20个产品,现仓储量为10
已经生产了10个产品,现仓储量为20
已经生产了10个产品,现仓储量为30
已经生产了10个产品,现仓储量为40
------要生产的产品数量80超过剩余库存量60,暂时不能执行生产任务!
-----准备消费50,当前剩余40,暂不能消费!

 

情况2:没有死锁的情况

-----准备消费50,当前剩余30,暂不能消费!
已经消费了30个产品,现仓储量为0
-----准备消费50,当前剩余0,暂不能消费!
已经生产了10个产品,现仓储量为10
已经生产了80个产品,现仓储量为90
已经生产了10个产品,现仓储量为100
------要生产的产品数量10超过剩余库存量0,暂时不能执行生产任务!
------要生产的产品数量10超过剩余库存量0,暂时不能执行生产任务!
------要生产的产品数量10超过剩余库存量0,暂时不能执行生产任务!
------要生产的产品数量10超过剩余库存量0,暂时不能执行生产任务!
已经消费了20个产品,现仓储量为80
已经生产了10个产品,现仓储量为90
已经生产了10个产品,现仓储量为100
------要生产的产品数量10超过剩余库存量0,暂时不能执行生产任务!
------要生产的产品数量10超过剩余库存量0,暂时不能执行生产任务!
已经消费了50个产品,现仓储量为50
已经生产了10个产品,现仓储量为60
已经生产了10个产品,现仓储量为70

 

上面代码中标黄的两个while能否换成 if 呢?如果我们换成if,可能得到如下的结果:

已经消费了20个产品,现仓储量为10
已经生产了10个产品,现仓储量为20
已经生产了10个产品,现仓储量为30
已经生产了10个产品,现仓储量为40
------要生产的产品数量80超过剩余库存量60,暂时不能执行生产任务!
-----准备消费50,当前剩余40,暂不能消费!
已经消费了30个产品,现仓储量为10
已经生产了10个产品,现仓储量为20
已经消费了50个产品,现仓储量为-30
已经生产了10个产品,现仓储量为-20
已经生产了80个产品,现仓储量为60
已经生产了10个产品,现仓储量为70

 

显然,换成if后,可能会出现非预期的效果,通过这个比较,我们可以发现,线程调用wait()又被唤醒后,是会继续执行wait()之后的代码的。

而我们的业务要求是,当前的生产数量一定不能大于剩余库存  且  当前准备消费的数量 一定不能小于库存,所以当线程被唤醒后,需要再次做出库存的校验,所以代码中的while一定不能换成if。

 

还有一个问题,为什么执行的时候,有时会出现死锁的情况呢?(出现死锁时,我们可以发现eclipse的控制台上的运行的红色按钮始终没有熄灭,说明程序还没有执行完,这时,不管过多久程序都不会结束了,因为,程序死锁了)

我们分析一下最后输出的日志:

------要生产的产品数量80超过剩余库存量60,暂时不能执行生产任务!
-----准备消费50,当前剩余40,暂不能消费!

我们可以看出,当producer线程准备去生产的时候,发现剩余库存不足,所以它调用wait(),放弃了对象锁(这里的对象锁为Producer的实例p7),处于等待状态。

而此时consumer线程准备去消费的时候,发现库存不足,不能消费,所以它调用wait(),放弃对象锁(c1),处于等待状态。

这两个线程相互等待,最终也没有线程去唤醒它们,所以出现了死锁。

 

 

最后再贴一个死锁的代码:

/** * Java线程:并发协作-死锁 *  */public class Test {    public static void main(String[] args) {        DeadlockRisk dead = new DeadlockRisk();        MyThread t1 = new MyThread(dead, 1, 2);        MyThread t2 = new MyThread(dead, 3, 4);        MyThread t3 = new MyThread(dead, 5, 6);        MyThread t4 = new MyThread(dead, 7, 8);        t1.start();        t2.start();        t3.start();        t4.start();    }}class MyThread extends Thread {    private DeadlockRisk dead;    private int a, b;    MyThread(DeadlockRisk dead, int a, int b) {        this.dead = dead;        this.a = a;        this.b = b;    }    @Override    public void run() {        dead.read();        dead.write(a, b);    }}class DeadlockRisk {    private static class Resource {        public int value;    }    private Resource resourceA = new Resource();    private Resource resourceB = new Resource();    public int read() {        synchronized (resourceA) {            System.out.println("read():" + Thread.currentThread().getName() + "获取了resourceA的锁!");            synchronized (resourceB) {                System.out.println("read():" + Thread.currentThread().getName() + "获取了resourceB的锁!");                return resourceB.value + resourceA.value;            }        }    }    public void write(int a, int b) {        synchronized (resourceB) {            System.out.println("write():" + Thread.currentThread().getName() + "获取了resourceA的锁!");            synchronized (resourceA) {                System.out.println("write():" + Thread.currentThread().getName() + "获取了resourceB的锁!");                resourceA.value = a;                resourceB.value = b;            }        }    }}
View Code

并非每次执行都会死锁,要多试几次。

线程发生死锁可能性很小,即使看似可能发生死锁的代码,在运行时发生死锁的可能性也是小之又小。

发生死锁的原因一般是两个对象的锁相互等待造成的。

上面的代码出现死锁时控制台如下:

并发协作:多线程中的生产者与消费者模型