生产者消费者

视频地址：https://www.bilibili.com/video/BV1Jx411a7Dp

参考链接：https://blog.csdn.net/ncepu_Chen/article/details/95538587

sychronized+wait()+notifyAll()

// 产品类
class Product {
    int id;

    Product(int id) {
        this.id = id;
    }

    @Override
    public String toString() {
        return "Product{id=" + id + '}';
    }
}

// 同步栈,用于在生产者线程和消费者线程之间通信
class SyncStack {
    int index = 0;  // 栈顶元素上一位的下标

    Product[] products = new Product[4];

    // 向栈中送入产品,synchronized方法保证原子性
    public synchronized void push(Product product) {
        // 栈满了,停止生产
        while (index == products.length) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 栈不满,通知消费者线程消费
        // 因为notify()方法不能指定唤醒哪个线程,若只唤醒了另一生产者线程,则会发生死锁,因此我们需要把所有线程都唤醒.
        this.notifyAll();
        products[index] = product;
        index++;
    }

    // 从栈中取出产品,synchronized保证原子性
    public synchronized Product pop() {
        // 栈空了,停止消费
        while (index == 0) {
            // 为防止wait时发生异常后执行程序剩余部分或栈被其它消费者线程消费空,使用while而非if检测栈状态
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 栈不空,通知生产者线程生产
        // 因为notify()方法不能指定唤醒哪个线程,若只唤醒了另一消费者线程,则会发生死锁,因此我们需要把所有线程都唤醒.
        this.notifyAll();
        index--;
        return products[index];
    }
}

// 生产者类
class Producer implements Runnable {
    // 生产者工作的栈
    SyncStack syncStack;

    public Producer(SyncStack syncStack) {
        this.syncStack = syncStack;
    }

    @Override
    public void run() {
        // 生产10个产品
        for (int i = 0; i < 10; i++) {
            Product product = new Product(i);
            syncStack.push(product);
            System.out.println("produce" + product);
            try {
                Thread.sleep((int)(Math.random()*200));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

// 消费者类
class Consumer implements Runnable {
    // 消费者消费的栈
    SyncStack syncStack;

    public Consumer(SyncStack syncStack) {
        this.syncStack = syncStack;
    }

    @Override
    public void run() {
        // 消费10个产品
        for (int i = 0; i < 10; i++) {
            Product product = syncStack.pop();
            System.out.println("consume" + product);
            try {
                Thread.sleep((int)(Math.random()*1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public class ProducerConsumer {

    public static void main(String[] args) {
        SyncStack syncStack = new SyncStack();
        // 生产者和消费者都注入了同一个同步栈,线程间的锁都锁在同步栈syncStack上
        Producer producer = new Producer(syncStack);    // 生产者
        Consumer consumer = new Consumer(syncStack);    // 消费者
        // 创建多个生产者线程和消费者线程
        new Thread(producer, "生产者1").start();    // 生产者线程1    
        new Thread(producer, "生产者2").start();    // 生产者线程2
        new Thread(consumer, "消费者1").start();    // 消费者线程1
        new Thread(consumer, "消费者2").start();    // 消费者线程2
    }
}

Condition+ReentrantLock

// 产品类
class Product {
    int id;

    public Product(int id) {
        this.id = id;
    }

    @Override
    public String toString() {
        return "Product{id=" + id + '}';
    }
}

// 同步栈,用于在生产者线程和消费者线程之间通信
class SyncStack {

    private Lock lock = new ReentrantLock();        // 锁对象

    // 绑定在锁上的一个条件,阻塞在该条件上的线程为生产者线程
    private Condition producerCondition = lock.newCondition();

    // 绑定在锁上的一个条件,阻塞在该条件上的线程为消费者线程
    private Condition consumerCondition = lock.newCondition();

    int MAX_SIZE = 5;      // 同步队列的最大容量
    Queue<Product> products = new LinkedList<>();   // 同步队列


    // 向栈中送入产品
    public void push(Product product) {
        lock.lock();
        try {
            while (products.size() == MAX_SIZE) {
                try {
                    // 当前线程应该是生产者线程,因此将当前线程阻塞在producerCondition上
                    producerCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            products.add(product);
            // 唤醒所有阻塞在consumerCondition的线程,这些被唤醒的线程都应该是消费者线程
            consumerCondition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    // 从栈中取出产品
    public Product pop() {
        lock.lock();
        try {
            while (products.size() == 0) {
                try {
                    // 当前线程应该是消费者线程,因此将当前线程阻塞在consumerCondition上
                    consumerCondition.await();
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            Product product = products.peek();
            products.remove();
            // 唤醒所有阻塞在producerCondition的线程,这些被唤醒的线程都应该是生产者线程
            producerCondition.signalAll();
            return product;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        SyncStack syncStack = new SyncStack();

        // 启动2个生产者线程
        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                // 生产10个产品
                for (int j = 0; j < 10; j++) {
                    Product product = new Product(j);
                    syncStack.push(product);
                    System.out.println(Thread.currentThread().getName() + "produce" + product);
                    try {
                        Thread.sleep((int) (Math.random() * 100));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, "生产者线程" + i).start();
        }

        // 启动2个消费者线程
        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                // 消费10个产品
                for (int j = 0; j < 10; j++) {
                    Product product = syncStack.pop();
                    System.out.println(Thread.currentThread().getName() + "consume" + product);
                    try {
                        Thread.sleep((int) (Math.random() * 1000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, "消费者线程" + i).start();
        }
    }
}