使用 BlockingQueue 实现生产者消费者模式

public class ProducerConsumerModel2 {

    private static BlockingQueue queue = new ArrayBlockingQueue<>(5);public static void main(String[] args) throws InterruptedException {        CountDownLatch countDownLatch = new CountDownLatch(100);        Runnable producer = () -> {try {                queue.put("数据");                System.out.println("生产数据...");            } catch (InterruptedException e) {                e.printStackTrace();            } finally {                countDownLatch.countDown();            }        };        Runnable consumer = () -> {try {                String product = queue.take();                System.out.println("消费" + product + "..");            } catch (InterruptedException e) {                e.printStackTrace();            } finally {                countDownLatch.countDown();            }        };        IntStream.rangeClosed(1, 50).forEach(i -> {            CompletableFuture.runAsync(producer);            CompletableFuture.runAsync(consumer);        });        countDownLatch.await();    }}

使用 wait 和 notify 实现生产者消费者模式

public class ProducerConsumerModel {    public static void main(String[] args) {        EventStorage eventStorage = new EventStorage();        Producer producer = new Producer(eventStorage);        Consumer consumer = new Consumer(eventStorage);

        new Thread(producer).start();        new Thread(consumer).start();    }

    static class Producer implements Runnable {        private EventStorage storage;

        public Producer(EventStorage storage) {            this.storage = storage;        }

        @Override        public void run() {            for (int i = 0; i 100; i++) {                storage.put(i);            }        }    }

    static class Consumer implements Runnable {        private EventStorage storage;

        public Consumer(EventStorage storage) {            this.storage = storage;        }

        @Override        public void run() {            for (int i = 0; i 100; i++) {                storage.take(i);            }        }    }

    static class EventStorage {        private int size;        private LinkedList storage;public EventStorage() {            size = 10;            storage = new LinkedList<>();        }public synchronized void put(int i) {while (storage.size() == this.size) {try {                    wait();                } catch (InterruptedException e) {                    e.printStackTrace();                }            }            storage.push("数据" + i);            System.out.println("生产 -> 数据" + i + " 当前仓库中已经有了 " + storage.size() + " 条数据");            notify();        }public synchronized void take(int i) {while (storage.size() == 0) {try {                    wait();                } catch (InterruptedException e) {                    e.printStackTrace();                }            }            System.out.println("消费 -> " + storage.poll() + " 当前仓库中剩下 " + storage.size() + " 条数据");            notify();        }    }}

源码地址：https://github.com/mirrormingzZ/JavaInterview