Java基于阻塞队列实现生产者消费者模型示例详解

import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.LinkedBlockingQueue;
// 生产消费者模型——阻塞队列
public class Demo20 {
    public static void main(String[] args) {
        // 创建一个阻塞队列来作为交易场所
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);
        Thread t1 = new Thread(() -> {
            int count = 0;
            while(true) {
                try {
                    queue.put(count);
                    System.out.println("生产元素:" + count);
                    count++;
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        Thread t2 = new Thread(() -> {
            while(true) {
                while(true) {
                    try {
                        Integer n = queue.take();
                        System.out.println("消费元素:" + n);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        t1.start();
        t2.start();
    }
}

class MyBlockQueue {
    // 使用string类型的数组来保存元素，我们假设这里只存string
    private String[] items = new String[1000];
    //head表示指向队列的头部
    volatile private int head = 0;
    volatile private int tail = 0;
    volatile private int size = 0; // size表示元素个数
    private Object locker = new Object();
    public void put(String elem) throws InterruptedException {
        synchronized(locker) {
            while(size >= items.length) {
                //队列已满
                locker.wait();
                //return;
            }
            items[tail] = elem;
            tail++;
            if(tail >= items.length) {
                tail = 0;
            }
            //tail++和下面的if判断可以替换成tail = (tail + 1) % (items.length)
            //但是站在CPU的角度来看,其实还是简单的if判断比较快
            size++;
            locker.notify(); // 用来唤醒队列为空的阻塞情况
        }
    }
    //出队列
    public String take() throws InterruptedException {
        synchronized(locker) {
            while(size == 0) {
                locker.wait();
            }
            String elem = items[head];
            head++;
            if(head >= items.length) {
                head = 0;
            }
            size--;
            //使用notify来唤醒队列阻塞满的情况
            locker.notify();
            return elem;
        }
    }
}
public class Demo21 {
    public static void main(String[] args) {
        // 创建两个线程分别表示消费者和生产者
        MyBlockQueue queue = new MyBlockQueue();
        Thread t1 = new Thread(() -> {
           int count = 0;
           while(true) {
               try {
                   queue.put(count + "");
                   System.out.println("生产元素: " + count);
                   count++;
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
        });
        Thread t2 = new Thread(() -> {
            while(true) {
                try {
                    String count = queue.take();
                    System.out.println("消费元素: " + count);
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        t1.start();
        t2.start();
    }
}