生产者消费者设计模式

港控/mmm° 2022-11-29 13:27 172阅读 0赞

## 生产者消费者设计模式 ##

概述

> 有两个线程生产者和消费者，有一个任务队列（有大小限制，太大了阻塞当前的生产者，不让其生产）。队列中有put和take方法。
> 
> put方法时将消息(Message)放入队列中，唤醒消费者进行消费
> 
> take拿出一个消息进行消费。唤醒生产者可以生产。

代码实现如下

1.编写Message消息类

public class Message { 
        private String data;
    
        public Message(String data) { 
            this.data = data;
        }
    
        public String getData(){ 
            return this.data;
        }
    }

2.创建消息队列

消息队列的take方法和put方法在出现异常时是break？还是return？

分析：个人认为在模拟代码中出现异常只能为中端此线程interrupt()，线程中断了这时不需要在put了直接返回就行了。

public class MessageQueue { 
        private final LinkedList<Message> queue = new LinkedList<>();
    
        private static final int DEFAULT_QUEUE_SIZE = 100;
    
        private final int queueSize;
    
        public MessageQueue(){ 
            this(DEFAULT_QUEUE_SIZE);
        }
    
        public MessageQueue(int queueSize) { 
            this.queueSize = queueSize;
        }
    
        public void put(Message message){ 
            synchronized (queue){ 
                while(queue.size() > queueSize){ 
                    try { 
                        queue.wait();
                    } catch (InterruptedException e) { 
                        return;
                    }
                }
                queue.addLast(message);
                queue.notifyAll();
            }
        }
    
        public Message take(){ 
            synchronized (queue){ 
                while(queue.isEmpty()){ 
                    try { 
                        queue.wait();
                    } catch (InterruptedException e) { 
                        return null;
                    }
                }
                Message message = queue.removeFirst();
                queue.notifyAll();
                return message;
            }
        }
    }

3.模拟生产者

public class ProducerThread extends Thread { 
        private final MessageQueue queue;
    
        private static final AtomicInteger counter = new AtomicInteger(0);
    
        public ProducerThread(MessageQueue queue,int seq) { 
            super("producer - "+seq);
            this.queue = queue;
        }
    
        @Override
        public void run() { 
            while(true){ 
                Message message = new Message(String.valueOf(counter.getAndIncrement()));
                queue.put(message);
                System.out.println(getName() + " 当前生产的数据为 ： "+message.getData());
                try { 
                    Thread.sleep(2000);
                } catch (InterruptedException e) { 
                    break;
                }
            }
        }
    }

4.模拟消费者

public class ConsumerThread extends Thread { 
        private final MessageQueue queue;
    
        public ConsumerThread(MessageQueue queue,int seq) { 
            super("consumer - "+seq);
            this.queue = queue;
        }
    
        @Override
        public void run() { 
            while(true){ 
                Message message = queue.take();
                System.out.println(getName() + " 当前消费的数据: "+message.getData());
                try { 
                    Thread.sleep(3000);
                } catch (InterruptedException e) { 
                    break;
                }
            }
        }
    }

5.测试代码

public class Application { 
        private static final MessageQueue QUEUE = new MessageQueue();
        public static void main(String[] args) throws InterruptedException { 
            new ConsumerThread(QUEUE,1).start();
            new ConsumerThread(QUEUE,2).start();
    
    
            Thread.sleep(2000);
            new ProducerThread(QUEUE,1).start();
            new ProducerThread(QUEUE,2).start();
    
        }
    }