手写生产者消费者模型

前言

生产者-消费者模式是一个十分经典的多线程并发协作模式，弄懂生产者-消费者问题能够让我们对并发编程的理解加深。

所谓的生产者-消费者，实际上包含了两类线程，一种是生产者线程用于生产数据，另一种是消费者线程用于消费数据，为了解耦生产者和消费者的关系，通常会采用共享的数据区域，就像是一个仓库，生产者生产数据之后直接放置在共享数据区中，并不需要关心消费者的行为；而消费者只需要从共享数据区中获取数据，不需要关心生产者的行为。

这个共享数据区域中应该具备这样的线程间并发协作功能：

1. 如果共享数据区已满的话，阻塞生产者继续生产数据；
2. 如果共享数据区为空的话，阻塞消费者继续消费数据；

在实现生产者消费者问题时，可以采用三种方式：

1. 使用 BlockingQueue 实现
2. 使用 synchronized以及Object wait/notify 的消息通知机制；
3. 使用 Lock Condition 的 await/signal 消息通知机制；

BlockingQueue 实现生产者-消费者

BlockingQueue 提供了可阻塞的插入和移除的方法。当队列容器已满，生产者线程会被阻塞，直到队列未满；当队列容器为空时，消费者线程会被阻塞，直至队列非空时为止。

有了这个队列，生产者就只需要关注生产，而不用管消费者的消费行为，更不用等待消费者线程执行完；消费者也只管消费，不用管生产者是怎么生产的，更不用等着生产者生产。

public class ProductorConsumer {

    private static LinkedBlockingQueue queue = new LinkedBlockingQueue<>();

    public static void main(String[] args) {
        ExecutorService service = Executors.newFixedThreadPool(15);
        for (int i = 0; i < 5; i++) {
            service.submit(new Productor(queue));
        }
        for (int i = 0; i < 10; i++) {
            service.submit(new Consumer(queue));
        }
    }


    static class Productor implements Runnable {

        private BlockingQueue queue;

        public Productor(BlockingQueue queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                while (true) {
                    Random random = new Random();
                    int i = random.nextInt();
                    System.out.println("生产者" + Thread.currentThread().getName() + "生产数据" + i);
                    queue.put(i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Consumer implements Runnable {
        private BlockingQueue queue;

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

        @Override
        public void run() {
            try {
                while (true) {
                    Integer element = (Integer) queue.take();
                    System.out.println("消费者" + Thread.currentThread().getName() + "正在消费数据" + element);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

synchronized 实现生产者-消费者

这其实也是手动实现阻塞队列的方式

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.CyclicBarrier;

public class MyBlockingQueue {

    //队列
    private final Queue myQueue = new LinkedList<>();

    //最大长度
    private static final int MAXSIZE = 20;
    private static final int MINSIZE = 0;

    //获取队列长度
    public int getSize() {
        return myQueue.size();
    }

    //生产者
    public void push(String str) throws Exception {
        //拿到对象锁
        synchronized (myQueue) {

            //如果队列满了，则阻塞
            while (getSize() == MAXSIZE) {
                myQueue.wait();
            }

            myQueue.offer(str);
            System.out.println(Thread.currentThread().getName() + "放入元素" + str);

            //唤醒消费者线程，消费者和生产者自己去竞争锁
            myQueue.notify();
        }
    }

    //消费者
    public String pop() throws Exception {
        synchronized (myQueue) {
            String result = null;

            //队列为空则阻塞
            while (getSize() == MINSIZE) {
                myQueue.wait();
            }
            //先进先出
            result = myQueue.poll();

            System.out.println(Thread.currentThread().getName() + "取出了元素" + result);
            //唤醒生产者线程，消费者和生产者自己去竞争锁
            myQueue.notify();

            return result;
        }
    }

    public static void main(String args[]) {

        MyBlockingQueue myBlockingQueue = new MyBlockingQueue();

        //两个线程，都执行完成了打印
        CyclicBarrier barrier = new CyclicBarrier(2, () -> {
            System.out.println("生产结束，下班了，消费者明天再来吧！");
        });

        //生产者线程
        new Thread(() -> {
            //50个辛勤的生产者循环向队列中添加元素
            try {
                for (int i = 0; i < 50; i++) {
                    myBlockingQueue.push("——" + i);
                }
                //生产完了
                barrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "生产者").start();

        //消费者线程
        new Thread(() -> {
            //50个白拿的消费者疯狂向队列中获取元素
            try {
                for (int j = 0; j < 50; j++) {
                    myBlockingQueue.pop();
                }
                //消费完了
                barrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "消费者").start();

    }
}

Condition 实现生产者-消费者

public class BoundedQueue {

    /**
     * 生产者容器
     */
    private LinkedList