Java实现生产消费者模型

迷南。 2022-09-03 05:09 150阅读 0赞

# 1. 使用BlockingQueue #

Java内置了许多BlockingQueue的实现，例如ArrayBlockingQueue，LinkedBlockingQueue等。我们可以通过这些内置的阻塞队列来实现生产者消费者模型。

下面是producer的实现：

public class Producer implements Runnable {
        private String name;
        private BlockingQueue<String> queue;
    
        public Producer(String name, BlockingQueue<String> queue) {
            this.name = name;
            this.queue = queue;
        }
    
        @Override
        public void run() {
            for (int i = 0; i < 100; i++) {
                try {
                    queue.put("message " + i);
                    System.out.println(name + " produces message " + i);
                    TimeUnit.SECONDS.sleep(1); // producer以每秒1条的速率生产100条消息
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

下面是consumer的实现：

public class Consumer implements Runnable {
        private String name;
        private BlockingQueue<String> queue;
    
        public Consumer(String name, BlockingQueue<String> queue) {
            this.name = name;
            this.queue = queue;
        }
    
        @Override
        public void run() {
            while (true) {
                try {
                    String message = queue.take();
                    System.out.println(name + " consumes " + message);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

在Main类中启动2个producer和1个consumer：

public class Main {
        public static void main(String[] args) {
            BlockingQueue<String> queue = new ArrayBlockingQueue<String>(5);
            new Thread(new Producer("producer1", queue)).start();
            new Thread(new Producer("producer2", queue)).start();
            new Thread(new Consumer("consumer1", queue)).start();
        }
    }

# 2. 使用wait/notify实现 #

我们也可以通过wait/notify机制实现一个简单的BlockingQueue，从而实现生产者消费者模型。

下面是wait/notify机制实现的SimpleBlockingQueue：

public class SimpleBlockingQueue<T> {
        private LinkedList<T> elements;
        private int capacity;
        private final Object lock = new Object();
    
        public SimpleBlockingQueue(int size) {
            elements = new LinkedList<>();
            this.capacity = size;
        }
    
        public void put(T element) throws InterruptedException {
            synchronized (lock) {
                while (elements.size() == capacity) {
                    lock.wait();
                }
                elements.add(element);
                lock.notifyAll();
            }
        }
    
        public T take() throws InterruptedException {
            synchronized (lock) {
                while (elements.isEmpty()) {
                    lock.wait();
                }
                T first = elements.pollFirst();
                lock.notifyAll();
                return first;
            }
        }
    }

Producer和Consumer的代码可以保持不变，只需要在构造时传入一个SimpleBlockingQueue实例。

# 3. 使用lock/condition实现 #

通过Lock锁，我们也能实现等待/通知模式。

下面是使用condition.await/condition.signal机制实现的SimpleBlockingQueue：

public class SimpleBlockingQueue<T> {
        private LinkedList<T> elements;
        private int capacity;
        private final Lock lock = new ReentrantLock();
        private final Condition condition = lock.newCondition();
    
        public SimpleBlockingQueue(int size) {
            elements = new LinkedList<>();
            this.capacity = size;
        }
    
        public void put(T element) throws InterruptedException {
            lock.lock();
            try {
                while (elements.size() == capacity) {
                    condition.await();
                }
                elements.add(element);
                condition.signalAll();
            } finally {
                lock.unlock();
            }
        }
    
        public T take() throws InterruptedException {
            lock.lock();
            try {
                while (elements.isEmpty()) {
                    condition.await();
                }
                T first = elements.pollFirst();
                condition.signalAll();
                return first;
            } finally {
                lock.unlock();
            }
        }
    }

我们也可以通过两个单独的condition开关来实现细粒度的等待/通知模型

public class SimpleBlockingQueue<T> {
        private LinkedList<T> elements;
        private int capacity;
        private final Lock lock = new ReentrantLock();
        private final Condition notFull = lock.newCondition();
        private final Condition notEmpty = lock.newCondition();
        
        public SimpleBlockingQueue(int size) {
            elements = new LinkedList<>();
            this.capacity = size;
        }
    
        public void put(T element) throws InterruptedException {
            lock.lock();
            try {
                while (elements.size() == capacity) {
                    notFull.await();
                }
                elements.add(element);
                notEmpty.signalAll();
            } finally {
                lock.unlock();
            }
        }
    
        public T take() throws InterruptedException {
            lock.lock();
            try {
                while (elements.isEmpty()) {
                    notEmpty.await();
                }
                T first = elements.pollFirst();
                notFull.signalAll();
                return first;
            } finally {
                lock.unlock();
            }
        }
    }