Java多线程基础：生产者/消费者模型

生产者/消费者模型

　　生产者消费者模型是多线程协作的经典模型，生产者线程负责产出数据，消费者线程负责消费生产者生产的数据，数据存放在共享区域内。该模型旨在合理的指导生产者和消费者进行生产或消费，避免过量生产以及无法消费的等问题。

　　 Java基础教程：多线程基础（3）——阻塞队列

　首先生产者和消费者互相解耦，那线程同步的重任放在了共享区域内：

当共享区域存满时，阻塞生产者，防止其继续生产。
当共享区域清空时，阻塞消费者，防止其继续消费。

说明：这里的阻塞并不等同指线程的阻塞态，实现阻塞的方式有很多种。　

线程调用sleep。

线程调用阻塞式IO方法，在该方法返回前，该线程被阻塞。

线程试图获得一个同步监视器，但该同步监视器正在被其他线程所持有。

线程等待某个通知。

程序调用了suspend方法将该线程挂起，非常容易导致死锁，避免即可。　

基于等待/唤醒机制实现模型

实现原理

　　object.wait()方法的作用是使当前执行代码的线程进入等待态，该方法用来将当前线程置入“预执行队列”，并在wait所在的代码行处停止执行，直到接到通知或被中断为止。wait方法会释放锁，所以它一定是在同步方法或同步代码块中使用。

　　obejct.notify()方法也是在同步方法或者同步块中使用，该方法用于通知那些可能等待该对象的对象锁的其他线程，如果有多个线程等待，则由线程规划器随机挑选出其中一个呈wait状态的线程，对其发出通知notify，并使它等待获取该对象的对象锁。

说明：在执行notify方法后，当前线程不会马上释放该对象锁，呈wait状态的线程也并不能马上获取该对象锁，要等到执行notify方法的线程将程序执行完，也就是退出Synchronized代码块后，才会释放锁，而呈wait状态所在的线程才可以获取该对象锁。

　　一句话总结就是：wait是线程停止运行，已达到阻塞的目的，而notify使停止的线程继续运行。它们两个互相配合，以实现线程间相互通信。

代码实现

生产者，负责生产数据。

class Producer extends Thread{

    private Buffer buffer;

    public Producer(Buffer buffer){

        this.buffer = buffer;

    }

    @Override

    public void run() {

       for (int i=0;i<10;i++){

           try {

               buffer.add(i);

               Thread.sleep(1000);

           } catch (InterruptedException e) {

               e.printStackTrace();

           }

       }

    }

}

消费者，负责消费数据。

class Consumer extends Thread{

    private Buffer buffer;

    public Consumer(Buffer buffer){

        this.buffer = buffer;

    }

    @Override

    public void run() {

        for(int i=0;i<10;i++){

            try {

                System.out.println(buffer.pull());

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

        }

    }

}

缓存区，即共享区域。

class Buffer{

    private Queue<Integer> list;

    private int size;

    public Buffer(int size) {

        list = new LinkedList<>();

        this.size = size;

    }

    public void add(int val) throws InterruptedException {

        synchronized (this){

            if(list.size()>=size)

                wait();

            list.add(val);

            notify();

        }

    }

    public int pull() throws InterruptedException {

        synchronized (this){

            if(list.size()==0)

                wait();

            int val = list.poll();

            notify();

            return val;

        }

    }

}

参考资料

《Java多线程编程艺术》

秒客网