【Java基础-并发编程】-异步模式之生产者/消费者

港控/mmm° 2023-10-12 21:44 1阅读 0赞

## 一、定义要点 ##

*  与前面的保护性暂停中的GuardedObject不同，不需要产生结果和消费结果的线程一一对应
 *  消费队列可以用来平衡生产和消费的线程资源
 *  生产者仅负责产生结果数据，不关心数据该如何处理，而消费者专心处理结果数据
 *  消息队列是有容量限制的，满时不会加入数据，空时不会再消耗数据
 *  JDK中各种阻塞队列，采用的就是这种模式

![90b40e2ef36142329da5b3328eb4bbfe.png][]

**代码示例**

public class TestMessageQueue {
        public static void main(String[] args)  {
            MessageQueue messageQueue = new MessageQueue(2);
            for (int i = 0; i < 3; i++) {
                int finalI = i;
                new Thread(()->{
                    messageQueue.put(new Message(finalI,"商品"+finalI));
                },"生产者"+i).start();
            }
    
            new Thread(()->{
                while (true){
                    try {
                        Thread.sleep(1000);
                        messageQueue.take();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            },"消费者").start();
        }
    }
    
    @Slf4j(topic = "c.MessageQueue")
    class MessageQueue{
    
        //消息队列的集合
        private LinkedList<Message> list = new LinkedList<>();
        //队列容量
        private int capcity;
    
        public MessageQueue(int capcity) {
            this.capcity = capcity;
        }
    
        //1.存入消息
        public void put(Message message){
            synchronized (list){
                while(list.size()==capcity){
                    log.info("队列已满，生产者线程等待");
                    try {
                        list.wait();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
                list.addLast(message);
                log.info("新的消息产生：{}",message);
                list.notifyAll();
            }
    
        }
        //2. 消费消息
        public Message take(){
            synchronized (list){
                while(list.size()==0){
                    log.info("队列已空，消费者线程等待");
                    try {
                        list.wait();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
                Message message = list.removeFirst();
                log.info("消费消息：{}",message);
                list.notifyAll();
                return message;
            }
        }
    
        //获取所有消息
    
        public LinkedList<Message> getList() {
            return list;
        }
    }
    
    final class Message{
        private int id;
        private Object value;
    
        public Message(int id, Object value) {
            this.id = id;
            this.value = value;
        }
    
        public int getId() {
            return id;
        }
    
        public Object getValue() {
            return value;
        }
    
        @Override
        public String toString() {
            return "Message{" +
                    "id=" + id +
                    ", value=" + value +
                    '}';
        }
    }

[90b40e2ef36142329da5b3328eb4bbfe.png]: https://img-blog.csdnimg.cn/90b40e2ef36142329da5b3328eb4bbfe.png