二.多线程设计模式篇-2.9 多线程Producer and Consumer设计模式

红太狼 2022-12-14 03:53 84阅读 0赞

## 1.问题的引入 ##

有一个工厂，他负责生产商品，他有很多工人，每天能生产出很多种类的商品，但是，他并不知道客户的消费情况，这也就造成了如果他产能过剩时，他将会有生产浪费的可能。

在Java程序中，若出现这种情况，则会产生内存的使用量过高，甚至有可能产生内存溢出发的可能。

## 2. 解决的方案 ##

利用中间生产的限额标定，即可解决产能过剩的情况。

## 3. 代码的实现 ##

(1)货物

/**
     * 货物
     */
    public class Production {
    
        private final String data;
    
        public Production(String data) {
            this.data = data;
        }
    
        public String getData() {
            return data;
        }
    }

(2)仓库

/**
     * 货物工厂（线程安全）
     */
    public class ProductionQueue {
        private final LinkedList<Production> productions;
        private final static int DEFAULT_MAX_PRODUCT_NUM = 100;
        private final int useLimit;
    
    
        public ProductionQueue() {
            this(DEFAULT_MAX_PRODUCT_NUM);
        }
    
        public ProductionQueue(int useLimit) {
            this.productions = new LinkedList<>();
            this.useLimit = useLimit;
        }
    
        public synchronized void put(Production production) throws InterruptedException {
            while (productions.size()>=useLimit){
                this.wait();
            }
            productions.addLast(production);
            this.notifyAll();
        }
    
        public synchronized Production take() throws InterruptedException {
            while (productions.isEmpty()){
                this.wait();
            }
            this.notifyAll();
            return productions.removeFirst();
        }
    
    }

(3)生产者

/**
     * 生产者
     */
    public class ProductThread extends Thread {
        private final ProductionQueue productionQueue;
        private static int count;
        private final Random random = new Random(System.currentTimeMillis());
    
        public ProductThread(ProductionQueue productionQueue, int seq) {
            super("PRODUCT"+seq);
            this.productionQueue = productionQueue;
        }
    
        @Override
        public void run() {
            while (true) {
                try {
                    System.out.println(Thread.currentThread().getName() + " set =>" + count);
                    productionQueue.put(new Production(Thread.currentThread().getName() + " set =>" + count++));
                    Thread.sleep(random.nextInt(1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

(4)消费者

/**
     * 消费者
     */
    public class ConsumerThread extends Thread {
        private final ProductionQueue productionQueue;
    
        private final Random random = new Random(System.currentTimeMillis());
    
        public ConsumerThread(ProductionQueue productionQueue, int seq) {
            super("CONSUMER"+seq);
            this.productionQueue = productionQueue;
        }
    
        @Override
        public void run() {
            while (true){
                try {
                    Production take  = productionQueue.take();
                    System.out.println(Thread.currentThread().getName()+" get =>"+take.getData());
                    Thread.sleep(random.nextInt(1000));
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
    }

(5)测试

/**
     * 测试生产消费
     */
    public class PCClient {
        public static void main(String[] args) throws InterruptedException {
            //存放仓库
            ProductionQueue productionQueue = new ProductionQueue();
            //生产者
            IntStream.range(0,4).forEach(value ->
            {
                ProductThread productThread = new ProductThread(productionQueue, value);
                productThread.setDaemon(true);
                productThread.start();
            });
            //消费者
            IntStream.range(0,5).forEach(value ->
            {
                ConsumerThread consumerThread = new ConsumerThread(productionQueue, value);
                consumerThread.setDaemon(true);
                consumerThread.start();
            });
    
            Thread.sleep(60_000L);
    
        }
    }