一、定义要点
- 与前面的保护性暂停中的GuardedObject不同,不需要产生结果和消费结果的线程一一对应
- 消费队列可以用来平衡生产和消费的线程资源
- 生产者仅负责产生结果数据,不关心数据该如何处理,而消费者专心处理结果数据
- 消息队列是有容量限制的,满时不会加入数据,空时不会再消耗数据
- JDK中各种阻塞队列,采用的就是这种模式
代码示例
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 +
'}';
}
}