生产者消费者模型java实现

系统管理员 2021-10-01 10:50 395阅读 0赞

方法一：利用synchronized，wait()和notify()

public class WaitAndNotify {
        public static void main(String[] args) throws IOException
        {
            Food food = new Food();
            new Thread(new Consumer("消费者一", food)).start();
    //      new Thread(new Consumer("消费者二", food)).start();
    //      new Thread(new Consumer("消费者三", food)).start();
    //
            new Thread(new Producer("生产者一", food)).start();
    //      new Thread(new Producer("生产者二", food)).start();
    //      new Thread(new Producer("生产者三", food)).start();
    
        }
    }
    class Producer implements Runnable
    {
        private Food food;
        private String producerName;
    
        public Producer(String producerName, Food food)
        {
            this.producerName = producerName;
            this.food = food;
        }
    
        @Override
        public void run()
        {
            while (true){
                try{
                    food.produce(producerName);
                } catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
        }
    
    }
    
    class Consumer implements Runnable
    {
    
        private Food food;
        private String consumerName;
    
        public Consumer(String consumerName, Food food)
        {
            this.consumerName = consumerName;
            this.food = food;
        }
    
        @Override
        public void run()
        {
            while (true){
                try{
                    food.consume(consumerName);
                } catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
        }
    
    }
    
    class Food
    {
        private AtomicInteger foodNum = new AtomicInteger(0);
    
        public void produce(String s) throws InterruptedException
        {
            synchronized (this)
            {
                while (foodNum.get()==5)
                {
                    System.out.println(s+"box is full，size = " + foodNum);
                    this.wait();
                }
                foodNum.incrementAndGet();
                System.out.println(s+"produce success foodNum = " + foodNum);
                this.notifyAll();
            }
    
        }
    
        public void consume(String s) throws InterruptedException
        {
            synchronized (this)
            {
                while (foodNum.get()==0)
                {
                    System.out.println(s+"box is empty,size = " + foodNum);
                    this.wait();
                }
                foodNum.decrementAndGet();
                System.out.println(s+"consume success foodNum = " + foodNum);
                this.notifyAll();
            }
    
        }
    }

方法二：利用lock，condition，await()和signal()  
Condition接口：出现替代了Object中的wait notify notifyAll方法。将这些监视器方法单独进行了封装，变成Condition监视器对象。可以任意锁进行组合。

public class AwaitAndSignal
    {
        public static void main(String[] args) throws IOException
        {
            Food food = new Food();
            new Thread(new Consumer(food), "消费者一").start();
    //        new Thread(new Consumer(food), "消费者二").start();
    //        new Thread(new Consumer(food), "消费者三").start();
    
            new Thread(new Producer(food), "生产者一").start();
    //        new Thread(new Producer(food), "生产者二").start();
    //        new Thread(new Producer(food), "生产者三").start();
        }
    }
    
    class Producer implements Runnable
    {
        private Food food;
    
        public Producer(Food food)
        {
            this.food = food;
        }
    
        @Override
        public void run()
        {
    
            while(true)
            {
                food.produce();
            }
    
        }
    
    }
    
    class Consumer implements Runnable
    {
    
        private Food food;
    
        public Consumer(Food food)
        {
            this.food = food;
        }
    
        @Override
        public void run()
        {
    
            while(true)
            {
                food.consume();
            }
    
        }
    
    }
    
    class Food
    {
        private int foodNum = 0;
    
        private ReentrantLock lock = new ReentrantLock();
    
        private Condition condition = lock.newCondition();
    
        private final int MAX_NUM = 5;
    
        public void produce()
        {
            lock.lock();
            try
            {
                while (foodNum == MAX_NUM)
                {
                    System.out.println("box is full，size = " + foodNum);
                    condition.await();
                }
                foodNum++;
                System.out.println("produce success foodNum = " + foodNum);
                condition.signalAll();
            }
            catch(InterruptedException e)
            {
                e.printStackTrace();
            } finally
            {
                lock.unlock();
            }
    
        }
    
        public void consume()
        {
            lock.lock();
            try
            {
                while (foodNum == 0)
                {
                    System.out.println("box is empty,size = " + foodNum);
                    condition.await();
                }
                foodNum--;
                System.out.println("consume success foodNum = " + foodNum);
                condition.signalAll();
            }
            catch(InterruptedException e)
            {
                e.printStackTrace();
            } finally
            {
                lock.unlock();
            }
    
        }
    }

方法三：使用BlockingQueue阻塞队列

![5057999-303b55544f6694d7.png][]

被阻塞的情况主要有如下两种:  
当队列满了的时候进行入队列操作  
当队列空了的时候进行出队列操作  
阻塞队列实现的生产者消费者模型,这里我们使用take()和put()方法

public class Testxx {
        public static void main(String[] args)
        {
            final BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);
            Runnable producerRunnable = new Runnable()
            {
                int i = 0;
                public void run()
                {
                    while (true)
                    {
                        try
                        {
                            System.out.println("我生产了一个" + i++);
                            bq.put(i + "");
                            Thread.sleep(1000);
                        } 
                        catch (InterruptedException e)
                        {
                            e.printStackTrace();
                        }
                    }
                }
            };
            Runnable customerRunnable = new Runnable()
            {
                public void run()
                {
                    while (true)
                    {
                        try
                        {
                            System.out.println("我消费了一个" + bq.take());
                            Thread.sleep(3000);
                        } 
                        catch (InterruptedException e)
                        {
                            e.printStackTrace();
                        }
                    }
                }
            };
            Thread producerThread = new Thread(producerRunnable);
            Thread customerThread = new Thread(customerRunnable);
            producerThread.start();
            customerThread.start();
        }
    }

[5057999-303b55544f6694d7.png]: /images/20210811/ea01de863a644354b4ea88e80f7fc0b6.png