java多线程之生产者消费者队列

2024年2月3日 148次阅读

java多线程一般都会讲消费者-生产者模型

生产者与消费者模型中，要保证以下几点：
1 同一时间内只能有一个生产者生产
2 同一时间内只能有一个消费者消费
3 生产者生产的同时消费者不能消费
4 消息队列满时生产者不能继续生产
5 消息队列空时消费者不能继续消费

—————————————————-Message类

public class Message {

 

public static int id;

public String content;

public String getContent() {

 return content;

}

public void setContent(String content) {

 this.content = content;

}

public int getId() {

 return id;

}

public void setId(int id) {

 Message.id = id;

}

}

————————————————————————————————————-Queue类

import java.util.ArrayList;
import java.util.List;

public class Queue {
 List<Message> queue = new ArrayList<Message>();

 /** 队列中message对象的最大值,默认为5 */
 int maxMessageNum = 5;

 public synchronized void produce(Message message) {

  this.notifyAll();
  while (queue.size() == maxMessageNum) {
   System.out.println(Thread.currentThread().getName()
     + ”  队列满！等待中。。。”);
   try {
    this.wait();
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
  queue.add(message);
  System.out.println(Thread.currentThread().getName() + “正在生产”
    + message.getContent() + “。。。  ,当前个数:” + getCount());

 }

 public synchronized void consume() {

  this.notifyAll();
  while (queue.size() == 0) {
   System.out.println(Thread.currentThread().getName()
     + ”  队列空！等待中。。。”);
   try {
    System.out.println(“begin!”);
    wait();
    System.out.println(“end!”);
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }

  Message message = queue.get(0);
  queue.remove(0);
  System.out.println(Thread.currentThread().getName() + “正在消费”
    + message.getContent() + “。。。 ,当前个数: ” + getCount());

 }

 public synchronized int getCount() {
  return queue.size();
 }
}

———————————————————————————————————————————————–Test类

public class Test {

public static void main(String[] args) {

 Queue Q = new Queue();

 Producer wQ1 = new Producer(Q);
 Producer wQ2 = new Producer(Q);
 
 Consumer rQ1 = new Consumer(Q);
 Consumer rQ2 = new Consumer(Q);
 Consumer rQ3 = new Consumer(Q);
 
 Thread threadWQ1 = new Thread(wQ1, “thread-wQ1”);
 Thread threadWQ2 = new Thread(wQ2, “thread-wQ2”);

 Thread threadRQ1 = new Thread(rQ1, “thread-rQ1”);
 Thread threadRQ2 = new Thread(rQ2, “thread-rQ2”);
 Thread threadRQ3 = new Thread(rQ3, “thread-rQ3”);
 
 threadWQ1.start();
 threadWQ2.start();
 
 threadRQ1.start();
  threadRQ2.start();
  threadRQ3.start();
}
}

class Producer extends Thread {

private Queue queue;

Producer(Queue queue) {
 this.queue = queue;
}

public void run() {
 
 while (true) {
  Message message = new Message();
  message.setId(++Message.id);
  message.setContent(“food”+Message.id);
  queue.produce(message);
  try {
   sleep(100);
  } catch (Exception e) {
  }
 }

}
}

class Consumer extends Thread {
private Queue queue;

Consumer(Queue queue) {
 this.queue = queue;
}

public void run() {
 while (true) {
  queue.consume();
  try {
   sleep(100);
  } catch (Exception e) {
  }

 }
}
}

点赞