用Java寫一個生產者-消費者隊列

生產者消費者的模型作用

1. 通過平衡生產者的生產能力和消費者的消費能力來提升整個系統的運行效率，這是生產者消費者模型最重要的作用。
2. 解耦，這是生產者消費者模型附帶的作用，解耦意味着生產者和消費者之間的聯系少，聯系越少越可以獨自發展  

使用阻塞隊列來實現

package yunche.test.producer;

import java.util.Random;
import java.util.concurrent.BlockingQueue;

/**
 * @ClassName: Producer
 * @Description: 生產者
 * @author: yunche
 * @date: 2018/08/26
 */
public class Producer implements Runnable
{

    private final BlockingQueue<Integer> queue;

    public Producer(BlockingQueue q)
    {
        this.queue = q;
    }

    @Override
    public void run()
    {
        try
        {
            while(true)
            {
                //模擬耗時1s
                Thread.sleep(1000);
                queue.put(produce());
            }
        }
        catch (InterruptedException e)
        {
            e.printStackTrace();
        }
    }

    private int produce()
    {
        int n = new Random().nextInt(10000);
        System.out.println("Thread: " + Thread.currentThread().getName() + " produce: " + n);
        return n;
    }
}

package yunche.test.producer;

import java.util.concurrent.BlockingQueue;

/**
 * @ClassName: Consumer
 * @Description: 消費者
 * @author: yunche
 * @date: 2018/08/26
 */
public class Consumer implements Runnable
{
    private final BlockingQueue<Integer> queue;

    public Consumer(BlockingQueue q)
    {
        this.queue = q;
    }

    @Override
    public void run()
    {
        while (true)
        {
            try
            {
                //模擬耗時
                Thread.sleep(2000);
                consume(queue.take());
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }

    }

    private void consume(Integer n)
    {
        System.out.println("Thread:" + Thread.currentThread().getName() + " consume: " + n);
    }
}


package yunche.test.producer;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * @ClassName: Main
 * @Description: 測試類
 * @author: yunche
 * @date: 2018/08/26
 */
public class Main
{
    public static void main(String[] args)
    {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(100);
        Producer p = new Producer(queue);
        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);

        Thread producer = new Thread(p);
        producer.setName("生產者線程");
        Thread consumer1 = new Thread(c1);
        consumer1.setName("消費者1");
        Thread consumer2 = new Thread(c2);
        consumer2.setName("消費者2");

        producer.start();
        consumer1.start();
        consumer2.start();

    }
}

使用wait-notify來實現

package yunche.test.producer;

import java.util.LinkedList;
import java.util.Random;


/**
 * @ClassName: Producer
 * @Description: 生產者
 * @author: yunche
 * @date: 2018/08/26
 */
public class Producer implements Runnable
{

    private final LinkedList<Integer> list;

    /**
     * 緩沖區大小
     */
    private final int maxSize;

    public Producer(LinkedList list, int size)
    {
        this.list = list;
        maxSize =size;
    }

    @Override
    public void run()
    {
        try
        {
            while(true)
            {
                //模擬耗時1s
                Thread.sleep(1000);
                synchronized (list)
                {
                    if(list.size()==maxSize)
                    {
                        System.out.println("緩沖區已滿，正在等待消費者消費..." + System.currentTimeMillis());
                        list.wait();
                    }
                    else
                    {
                        list.add(produce());
                        list.notifyAll();
                    }
                }

            }
        }
        catch (InterruptedException e)
        {
            e.printStackTrace();
        }
    }

    private int produce()
    {
        int n = new Random().nextInt(10000);
        System.out.println("Thread: " + Thread.currentThread().getName() + " produce: " + n);
        return n;
    }
}



package yunche.test.producer;

import java.util.Date;
import java.util.LinkedList;

/**
 * @ClassName: Consumer
 * @Description: 消費者
 * @author: yunche
 * @date: 2018/08/26
 */
public class Consumer implements Runnable
{
    private final LinkedList<Integer> list;

    public Consumer(LinkedList list)
    {
        this.list = list;
    }

    @Override
    public  void run()
    {
        while (true)
        {
            try
            {
                synchronized(list)
                {
                    //模擬耗時
                    Thread.sleep(1000);
                    if(list.isEmpty())
                    {
                        System.out.println("緩沖區已空，正在等待生產者生產..." + System.currentTimeMillis() + Thread.currentThread().getName());
                        list.wait();
                    }
                    else
                    {
                        consume(list.poll());
                        list.notifyAll();
                    }
                }

            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }

    }

    private void consume(Integer n)
    {
        System.out.println("Thread:" + Thread.currentThread().getName() + " consume: " + n);
    }
}


package yunche.test.producer;

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

/**
 * @ClassName: Main
 * @Description: 測試類
 * @author: yunche
 * @date: 2018/08/26
 */
public class Main
{
    public static void main(String[] args)
    {
        LinkedList<Integer> list = new LinkedList<>();
        Producer p = new Producer(list, 10);
        Consumer c1 = new Consumer(list);
        Consumer c2 = new Consumer(list);

        Thread producer = new Thread(p);
        producer.setName("生產者線程");
        Thread consumer1 = new Thread(c1);
        consumer1.setName("消費者1");
        Thread consumer2 = new Thread(c2);
        consumer2.setName("消費者2");

        producer.start();
        consumer1.start();
        consumer2.start();

    }
}

參考資料

　　Java面試題