生产者-消费者模式

实现一

通过一个阻塞队列来实现生产者-消费者模式，生产者将数据放入队列中，消费者从队列中取出数据进行处理

public class ProducerConsumerBQ {
    private static final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);

    public static void main(String[] args) {
        // producer
        new Thread(() -> {
            try {
                for(int i = 1; i < 10; i ++) {
                    queue.put(i);
                    System.out.println("producer -> {}" + i);
                    Thread.sleep(500);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "Producer").start();

        // consumer
        new Thread(() -> {
            try {
                while(true) {
                    Integer item = queue.take();
                    System.out.println("consumer -> {}" + item);
                    Thread.sleep(1000);
                }
            } catch(InterruptedException e) {
                e.printStackTrace();
            }
        }, "consumer").start();
    }
}

实现二

通过自定义一个阻塞队列来实现生产者-消费者模式

• 一个锁来保证线程安全
• 两个 Condition 变量来实现生产者和消费者的等待和唤醒机制

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Objects;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ProducerConsumerBQ {

    static class BlockingQueue<T> {
        private final Deque<T> deque = new ArrayDeque<>();
        private final int capacity;
        private final ReentrantLock lock = new ReentrantLock();

        private final Condition notEmpty = lock.newCondition();
        private final Condition notFull = lock.newCondition();

        private boolean closed = false;

        public BlockingQueue(int capacity) {
            if (capacity <= 0) {
                throw new IllegalArgumentException("capacity must be > 0");
            }
            this.capacity = capacity;
        }

        // 放入元素；如果队列已关闭，则抛出异常
        public void put(T element) throws InterruptedException {
            Objects.requireNonNull(element, "element cannot be null");
            lock.lockInterruptibly();
            try {
                while (deque.size() == capacity) {
                    if (closed) {
                        throw new IllegalStateException("queue is closed");
                    }
                    notFull.await();
                }

                if (closed) {
                    throw new IllegalStateException("queue is closed");
                }

                deque.addLast(element);
                notEmpty.signal();
            } finally {
                lock.unlock();
            }
        }

        // 取出元素；如果队列已关闭且为空，则返回 null 表示结束
        public T take() throws InterruptedException {
            lock.lockInterruptibly();
            try {
                while (deque.isEmpty()) {
                    if (closed) {
                        return null;
                    }
                    notEmpty.await();
                }

                T value = deque.removeFirst();
                notFull.signal();
                return value;
            } finally {
                lock.unlock();
            }
        }

        // 关闭队列，唤醒所有等待中的生产者和消费者
        public void close() {
            lock.lock();
            try {
                closed = true;
                notEmpty.signalAll();
                notFull.signalAll();
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> bq = new BlockingQueue<>(3);

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    Thread.sleep(1000);
                    String item = String.valueOf(i);
                    bq.put(item);
                    System.out.println("生产: " + item);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("生产者被中断，停止生产");
            } finally {
                bq.close();
                System.out.println("生产者结束，队列已关闭");
            }
        }, "producer");

        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    String item = bq.take();
                    if (item == null) {
                        System.out.println("队列已关闭且无数据，消费者退出");
                        break;
                    }
                    System.out.println("消费: " + item);
                    Thread.sleep(3000);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("消费者被中断，停止消费");
            }
        }, "consumer");

        producer.start();
        consumer.start();

        producer.join();
        consumer.join();
        System.out.println("主线程结束");
    }
}