生产者消费者模式

介绍

一个系统中存在生产者和消费者两个角色，他们通过缓存区进行通信，缓存区大小固定且有两个线程，一个线程控制生产者生成数据并存入缓存区，另一个线程供消费者从缓存区中取对象并去使用数据。

优点

生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据，两者会相互等待对方完成（注意防止死锁出现）。

应用场景

1. 如网络访问框架可以采用该模式，请求相当于生产者，响应相当于消费者
2. 用户提交订单，如果量过大，进入堵塞队列，等待订单生成后返回结果，才能继续执行生成订单的操作。

实现方式

• wait和notify实现
• 阻塞队列实现LinkedBlockingQueue

代码示例

1.wait和notify实现

缓存区：

public class Storage {

    private final int MAX_SIZE = 5;
    
    LinkedList<String> list = new LinkedList<String>();
    
    public void produce(String producer){
        
        synchronized (list) {
            while (list.size() == MAX_SIZE) {
                try {
                    System.out.println("队列已满，等待消费者使用");
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            
            System.out.println("添加：" + producer);
            list.add(producer);
            list.notifyAll();
        }
    }
    
    
    public void consume(String producer){
        
        synchronized (list) {
            while (list.size() == 0) {
                try {
                    System.out.println("等待生产者生成");
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("消费" + producer);
            list.remove(producer);
            list.notifyAll();
        }
    }
}

客户端中的生产者和消费者

public class Client {

    public static void main(String[] args) {
        Storage storage = new Storage();
        Producer producer = new Producer(storage);
        Consumer consumer = new Consumer(storage);
        producer.start();
        consumer.start();

    }

    public static class Producer extends Thread {
        Storage storage;

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

        @Override
        public void run() {
            for (int i = 0; i < 15; i++) {
                storage.produce("生产" + i);
            }
        }
    }

    public static class Consumer extends Thread {
        Storage storage;

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

        @Override
        public void run() {
            for (int i = 0; i < 15; i++) {
                storage.consume("生产" + i);
            }
        }
    }
}

1.LinkedBlockingQueue

生产者

public class Producer implements Runnable {

    private volatile boolean isRunning = true;
    // 创建一个队列
    BlockingQueue<PcData> queue;
    AtomicInteger count = new AtomicInteger();

    public Producer(BlockingQueue<PcData> queue) {
        this.queue = queue;
    }

    public void run() {
        PcData data = null;
        Random r = new Random();
        try {
            while (isRunning) {
                Thread.sleep(1000);
                data = new PcData(count.getAndIncrement());
                System.out.println("添加的队列中");
                if(!queue.offer(data, 2, TimeUnit.SECONDS)){
                    System.out.println("添加的队列失败，等待中");
                }
                
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
    }

    public void stop() {
        isRunning = false;
    }

}

消费者

public class Consumer implements Runnable {

    private BlockingQueue<PcData> queue;

    public Consumer(BlockingQueue<PcData> queue) {
        this.queue = queue;
    }

    public void run() {
        Random r = new Random();
        try {
            while (true) {
                PcData data = queue.take();
                if (null != data) {
                    int re = data.getData() * data.getData();
                    System.out.println(data.toString() + re);
                    Thread.sleep(r.nextInt(5000));
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            e.printStackTrace();
        }
    }

}

客户端

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<PcData> queue = new LinkedBlockingDeque<PcData>(2);

        Producer prducer1 = new Producer(queue);
        Producer prducer2 = new Producer(queue);
        Producer prducer3 = new Producer(queue);

        Consumer c1 = new Consumer(queue);
        Consumer c2 = new Consumer(queue);
        Consumer c3 = new Consumer(queue);

        ExecutorService executor = Executors.newCachedThreadPool();

        executor.execute(prducer1);
        executor.execute(prducer2);
        executor.execute(prducer3);
        executor.execute(c1);
        executor.execute(c2);
        executor.execute(c3);
        Thread.sleep(10 * 1000);
        prducer1.stop();
        prducer2.stop();
        prducer3.stop();
        Thread.sleep(3000);
        executor.shutdown();
    }
}