java并发编程——异步模式之生产者/消费者

╰半橙微兮° 2022-12-16 06:24 197阅读 0赞

# 异步模式之生产者/消费者 #

模式篇：

并发编程——[同步模式之保护性暂停][Link 1]  
并发编程——[异步模式之生产者/消费者][Link 2]

# 概述 #

*  与前面的保护性暂停中的 GuardObject 不同，不需要产生结果和消费结果的线程一一对应
 *  消费队列可以用来平衡生产和消费的线程资源
 *  生产者仅负责产生结果数据，不关心数据该如何处理，而消费者专心处理结果数据
 *  消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据
 *  JDK 中各种阻塞队列，采用的就是这种模式  
    ![在这里插入图片描述][20201019112451215.png_pic_center]  
    说明，t1,t2,t3是生产者，t4是消费者，这里的t4不一定是一个，有可能是多个。

# 实现 #

可以用一个双向队列，一头存，一头取。

import lombok.extern.slf4j.Slf4j;
    import java.util.LinkedList;
    
    @Slf4j(topic = "c.Test21")
    public class Test21 { 
    
        public static void main(String[] args) { 
            MessageQueue queue = new MessageQueue(2);
    
            for (int i = 0; i < 3; i++) { 
                int id = i;
                new Thread(() -> { 
                    queue.put(new Message(id , "值"+id));
                }, "生产者" + i).start();
            }
    
            new Thread(() -> { 
                while(true) { 
                    sleep(1);
                    Message message = queue.take();
                }
            }, "消费者").start();
        }
    
    }
    
    // 消息队列类 ， java 线程之间通信
    @Slf4j(topic = "c.MessageQueue")
    class MessageQueue { 
        // 消息的队列集合
        private LinkedList<Message> list = new LinkedList<>();
        // 队列容量
        private int capcity;
    
        public MessageQueue(int capcity) { 
            this.capcity = capcity;
        }
    
        // 获取消息
        public Message take() { 
            // 检查队列是否为空
            synchronized (list) { 
                while(list.isEmpty()) { 
                    try { 
                        log.debug("队列为空, 消费者线程等待");
                        list.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                // 从队列头部获取消息并返回
                Message message = list.removeFirst();
                log.debug("已消费消息 {}", message);
                list.notifyAll();
                return message;
            }
        }
    
        // 存入消息
        public void put(Message message) { 
            synchronized (list) { 
                // 检查对象是否已满
                while(list.size() == capcity) { 
                    try { 
                        log.debug("队列已满, 生产者线程等待");
                        list.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                // 将消息加入队列尾部
                list.addLast(message);
                log.debug("已生产消息 {}", message);
                list.notifyAll();
            }
        }
    }
    
    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 +
                    '}';
        }
    }

测试，三个线程模拟三个生产者，一个线程模拟一个消费者。  
![在这里插入图片描述][20201019115948165.png_pic_center]

[Link 1]: https://blog.csdn.net/e891377/article/details/109152523
[Link 2]: https://blog.csdn.net/e891377/article/details/109157442
[20201019112451215.png_pic_center]: /images/20221123/94c4d2cb18e04e5d92596dafb8a5c4f7.png
[20201019115948165.png_pic_center]: /images/20221123/c9808a44b1864777a63500d8d8d4237b.png