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

  • A+
所属分类:JUC并发编程

黑马视频教程:https://www.bilibili.com/video/BV16J411h7Rd?p=107

1.定义

  • 消费队列可以用来平衡生产和消费之间的平衡
  • 生产者仅负责产生结果的数据,不关心数据如何处理,而消费者专心处理数据的结果
  • 消息队列容量有限为空消费者无法消费,满时生产者无法生产。
  • JDK中的各种阻塞队列采用的就是这种模式。

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

2.实现

代码注释详细

/** * 消息类 */
@Slf4j
class Message{
   

    private int id;
    private Object message;

    public Message(int id,Object message){
   
        this.id = id;
        this.message = message;
    }

    public int getId() {
   
        return id;
    }

    public Object getMessage() {
   
        return message;
    }

    @Override
    public String toString() {
   
        return "Message{" +
                "id=" + id +
                ", message=" + message +
                '}';
    }


}

/** * 消息队列 */
@Slf4j
class MessageQueue{
   

    /** * 存放消息的队列 */
    private LinkedList<Message> queue;
    /** * 消息队列的容量大小 */
    private Integer capacity;

    public MessageQueue(Integer capacity) {
   
        this.capacity = capacity;
        queue = new LinkedList<>();
    }

    /** * put消息message到消息队列中 * @param message */
    public void put(Message message){
   
        synchronized (queue){
   
            // 队列已满就继续等待
            while (capacity.equals(queue.size())){
   
                try {
   
                    log.debug("消息队列满了,无法继续添加消息到队列中");
                    queue.wait();
                } catch (InterruptedException e) {
   
                    e.printStackTrace();
                }
            }
            log.debug("添加消息{}到消息队列中",message);
            queue.addLast(message);
            // 唤醒队列为空情况下等待的调用take方法的线程
            queue.notifyAll();
        }
    }

    /** * 消费消息 * @return */
    public Message take(){
   
        synchronized (queue){
   
            // 队列为空没有消息,等待生产者生产消息到消费队列中
            while (queue.size() == 0){
   
                try {
   
                    log.debug("消息队列为空,无法消费,继续等待");
                    queue.wait();
                } catch (InterruptedException e) {
   
                    e.printStackTrace();
                }
            }
            Message message = queue.removeFirst();
            log.debug("消费消息{}",message);
            // 唤醒因队列为满的情况下调用put方法等待的线程
            queue.notifyAll();
            return message;
        }
    }

}

3.测试

测试代码

	/** * 测试 * @param args */
    public static void main(String[] args) {
   
        MessageQueue messageQueue = new MessageQueue(2);

        // 模拟三个生产者
        for (int i = 0; i < 3; i++) {
   
            int id = i;
            new Thread(()->{
   
                messageQueue.put(new Message(id,"消息" +id));
            },"生产者线程 t"+i).start();
        }

        // 一个消费者
        new Thread(()->{
   
            while (true){
   
                log.debug("1s后消费");
                try {
   
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
   
                    e.printStackTrace();
                }
                Message take = messageQueue.take();
            }
        },"消费者线程 t1").start();
    }

测试结果
异步模式之生产者/消费者

w3cjava

发表评论

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen: