介绍

• 生产者负责产生数据，将数据存入消息队列，不关系数据该如何处理，当消息队列满时不再生产数据
• 消费者负责从消息队列中取数据，对数据进行处理，当消息队列为空时，停止消费
• 消息队列用来平衡生产和消费的消息资源，消费队列有容量限制

使用场景

• JDK中各种阻塞队列

代码实现

主类

import java.util.LinkedList;
//异步生产者/消费者模式
public class ProducerConsumerModel {public static void main(String[] args) {//测试MessageQueue messageQueue = new MessageQueue(3); //容量为3的消息队列//创建三个生产者线程生产消息for (int i = 0; i < 3; i++) {int messageNum=i;new Thread(()->{Message message = new Message(messageNum, 	"消息"+messageNum);try {messageQueue.depositInto(message);} catch (InterruptedException e) {e.printStackTrace();}},"生产者线程"+i).start();}//创建一个消费者线程消费消息new Thread(()->{try {while (true)messageQueue.takeOut();} catch (InterruptedException e) {e.printStackTrace();}},"消费者线程").start();
}
}

消息队列类

class MessageQueue{//真正维护消息的队列
LinkedList<Message> list = new LinkedList<Message>();//消息队列的容量
int capacity = 0;public MessageQueue(int capacity) {this.capacity = capacity;
}//存入消息
public void depositInto(Message message) throws InterruptedException {synchronized (list){while(list.size()==this.capacity){//如果消息队列满，阻塞当前线程，将线程放入list对象所关联的Monitor的waitSet进行等待System.out.println("消息队列容量已达上限");list.wait();}//向消息队列中存入消息list.push(message);//打印System.out.println(Thread.currentThread().getName()+"存入消息："+message);//释放list对象锁之前 ，唤醒其他在WaitSet中等待的线程，让其都进入EntryList有资格去竞争锁list.notifyAll();}}
//取出消息
public Message takeOut() throws InterruptedException {synchronized (list){while (list.isEmpty()){//如果消息队列为空，阻塞当前线程，将该线程放入list对象所关联的Monitor的waitSet中System.out.println("消息队列为空");list.wait();}//从消息队列中取出数据Message message = list.remove();//打印System.out.println(Thread.currentThread().getName()+"取出消息："+message);//唤醒在waitSet中的线程到entryList中list.notifyAll();return message;}
}
}

消息类

//消息
final class Message{private int id;private String value;public Message(int id, String value) {this.id = id;this.value = value;
}public int getId() {return id;
}public String getValue() {return value;
}@Override
public String toString() {return "Message{" +"id=" + id +", value='" + value + '\'' +'}';
}
}

执行结果