Java：生产者消费者模式

---

 

Java中的生产者、消费者问题描述：

　　生产者-消费者（producer-consumer）问题, 也称作有界缓冲区（bounded-buffer）问题, 两个进程共享一个公共的固定大小的缓冲区(仓库). 其中一个是生产者, 用于将产品放入仓库: 另外一个是消费者, 用于从仓库中取出产品消费. 问题出现在当仓库已经满了, 而此时生产者还想向其中放入一个新的产品的情形, 其解决方法是让生产者此时进行等待, 等待消费者从仓库中取走了一个或者多个产品后再去唤醒它. 同样地, 当仓库已经空了, 而消费者还想去仓库消费产品, 此时也可以让消费者进行等待, 等待生产者放入一个或者多个产品时再唤醒它.

生产者、消费者问题需明确以下信息：

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

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

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

　　4、消费者发现仓储没产品可消费时候会通知生产者生产.

生产者消费者模式实现：

1.生产者类Pro：

　　1）继承Thread类；

　　2）提供Pro(Store)构造方法；

　　3）重写run()方法：调用store的producePro方法；

/**
 * Created by leng on 2017/7/18.
 */
public class Pro extends Thread {
    Store store;
    public Pro(Store store) {
        this.store = store;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "开始生产产品！");
        while (true) {
            store.producePro();
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

2.消费者类Con：

　　1）继承Thread类；

　　2）提供Con(Store)构造方法；

　　3）重写run()方法：调用store的consumePro方法；

/**
 * Created by leng on 2017/7/18.
 */
public class Con extends Thread{
    Store store;
    public Con(Store store) {
        this.store = store;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "开始消费产品！");
        while (true) {
            store.consumePro();
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

3.仓库类Store：

　　1）创建queue作为仓库；使用有容量的PriorityQueue<>(size)；

　　2）producePro方法：1.同步queue；2.队列满时wait；3.入队offer(1)，唤醒；

　　3）consumePro方法：1.同步queue；2.队列空时wait；3.出队poll()，唤醒；

/**
 * Created by leng on 2017/7/18.
 */
public class Store {

    private int queueSize = 10;
    private PriorityQueue<Integer> queue = new PriorityQueue<>(queueSize);

    public void producePro() {
        synchronized (queue) {
            
            while (queue.size() == queueSize) {
                try {
                    System.out.println("队列已满，等待空余空间！");
                    queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            
            queue.offer(1);
            System.out.println(Thread.currentThread().getName()+"向队列取中插入一个元素，队列剩余空间："+(queueSize-queue.size()));
            queue.notify();
        }

    }

    public void consumePro() {
        synchronized (queue) {

            while (queue.size() == 0) {
                try {
                    System.out.println("队列已空，等待生产商品！");
                    queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            queue.poll();
            System.out.println(Thread.currentThread().getName()+"从队列中取走一个元素，队列中剩余"+queue.size()+"个");
            queue.notify();
        }
    }
}

4）测试类：

　　新建仓库，新建Pro和Con；

public class ProducerConsumer{
    public static void main(String[] args) {
        Store store = new Store();
        Pro p1 = new Pro(store);
        Pro p2 = new Pro(store);
        Pro p3 = new Pro(store);
        Pro p4 = new Pro(store);
        
        Con c1 = new Con(store);
        Con c2 = new Con(store);

        p1.start();
        p2.start();
        p3.start();
        p4.start();
        
        c1.start();
        c2.start();
    }
}

测试结果：

Thread-2开始生产产品！
Thread-0开始生产产品！
Thread-1开始生产产品！
Thread-2向队列取中插入一个元素，队列剩余空间：9
Thread-3开始生产产品！
Thread-1向队列取中插入一个元素，队列剩余空间：8
Thread-0向队列取中插入一个元素，队列剩余空间：7
Thread-3向队列取中插入一个元素，队列剩余空间：6
Thread-4开始消费产品！
Thread-5开始消费产品！
Thread-4从队列中取走一个元素，队列中剩余3个
Thread-5从队列中取走一个元素，队列中剩余2个
Thread-2向队列取中插入一个元素，队列剩余空间：7
Thread-3向队列取中插入一个元素，队列剩余空间：6
Thread-0向队列取中插入一个元素，队列剩余空间：5
Thread-1向队列取中插入一个元素，队列剩余空间：4
Thread-4从队列中取走一个元素，队列中剩余5个
Thread-5从队列中取走一个元素，队列中剩余4个
Thread-2向队列取中插入一个元素，队列剩余空间：5
Thread-0向队列取中插入一个元素，队列剩余空间：4
Thread-1向队列取中插入一个元素，队列剩余空间：3
Thread-4从队列中取走一个元素，队列中剩余6个
Thread-3向队列取中插入一个元素，队列剩余空间：3
Thread-5从队列中取走一个元素，队列中剩余6个
Thread-2向队列取中插入一个元素，队列剩余空间：3
Thread-1向队列取中插入一个元素，队列剩余空间：2
Thread-3向队列取中插入一个元素，队列剩余空间：1
Thread-4从队列中取走一个元素，队列中剩余8个
Thread-0向队列取中插入一个元素，队列剩余空间：1
Thread-5从队列中取走一个元素，队列中剩余8个
Thread-3向队列取中插入一个元素，队列剩余空间：1
Thread-5从队列中取走一个元素，队列中剩余8个
Thread-0向队列取中插入一个元素，队列剩余空间：1
Thread-2向队列取中插入一个元素，队列剩余空间：0
Thread-4从队列中取走一个元素，队列中剩余9个
Thread-1向队列取中插入一个元素，队列剩余空间：0
Thread-5从队列中取走一个元素，队列中剩余9个
Thread-1向队列取中插入一个元素，队列剩余空间：0
Thread-4从队列中取走一个元素，队列中剩余9个
Thread-2向队列取中插入一个元素，队列剩余空间：0
队列已满，等待空余空间！
队列已满，等待空余空间！
Thread-5从队列中取走一个元素，队列中剩余9个
Thread-0向队列取中插入一个元素，队列剩余空间：0
队列已满，等待空余空间！
Thread-4从队列中取走一个元素，队列中剩余9个
Thread-2向队列取中插入一个元素，队列剩余空间：0
队列已满，等待空余空间！
队列已满，等待空余空间！
Thread-5从队列中取走一个元素，队列中剩余9个
Thread-1向队列取中插入一个元素，队列剩余空间：0
队列已满，等待空余空间！
队列已满，等待空余空间！
队列已满，等待空余空间！
Thread-4从队列中取走一个元素，队列中剩余9个
Thread-3向队列取中插入一个元素，队列剩余空间：0
队列已满，等待空余空间！
队列已满，等待空余空间！
队列已满，等待空余空间！
Thread-5从队列中取走一个元素，队列中剩余9个
Thread-4从队列中取走一个元素，队列中剩余8个
Thread-0向队列取中插入一个元素，队列剩余空间：1
Thread-1向队列取中插入一个元素，队列剩余空间：0
队列已满，等待空余空间！

---

ps：线程启动的方法：

继承Thread：

testClass extends Thread；new testClass().start;

        Pro p1 = new Pro(store);
        Pro p2 = new Pro(store);

        p1.start();
        p2.start();

实现Runnable：

testClass implements Runnable；new Thread(new testClass()).start;

一个Runnable对象多个线程：

        Pro p1 = new Pro(store);
        Thread t1 = new Thread(p1);
        Thread t2 = new Thread(p1);

        t1.start();
        t2.start();

多个Runnable对象，每个Runnable对象对应一个线程：

        Pro p1 = new Pro(store);
        Pro p2 = new Pro(store);

        Thread t1 = new Thread(p1);
        Thread t2 = new Thread(p2);

        t1.start();
        t2.start();

Q：有什么区别？