JavaはBlockingQueueで簡単なメッセージキュー(生産者と消費者モデル)を実現

学習過程は、メモとして記録されています.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class ProducerAndConsumer {
	public static void main(String[] args) throws InterruptedException {
		  ProducerAndConsumer pac = new ProducerAndConsumer();
		  //       10     
		  BlockingQueue<Message> blockingQueue = new ArrayBlockingQueue<Message>(10);
		  //            
		  Producer producer = pac.new Producer(blockingQueue);
		  //            1
		  Consumer c1 = pac.new Consumer(blockingQueue);
		  c1.setName("Tread-c1");
		  c1.start();
		  //            2
		  Consumer c2 = pac.new Consumer(blockingQueue);
		  c2.setName("Tread-c2");
		  c2.start();
		  for(int i = 0; i < 1000; i++){
		   //         
		   Message message = pac.new Message();
		   message.setBody(i+"  body");
		   message.setTopoic(i+"  topoic");
		   producer.sendMessage(message);
		   //Thread.sleep(500);
		}
	}
	
//   
 class Producer{
	  private BlockingQueue<Message> queue;
	  public Producer(BlockingQueue<Message> queue){
	  	this.queue = queue;
	  }
	  public void sendMessage(Message message){
	   	try {
	    		queue.put(message);
	   	} catch (InterruptedException e) {
	   		 e.printStackTrace();
	   	}
	  }
}

//   
 class Consumer extends Thread{
	  private BlockingQueue<Message> queue;
	  public Consumer(BlockingQueue<Message> queue){
	  	 this.queue = queue;
	  }
	  @Override
	  public void run() {
	   	while(true){
	   		 if(!queue.isEmpty()){
		    		try {
		     			//System.out.println("  ："+Thread.currentThread().getName()+"    topic  ："+queue.take().getBody());
		      			System.out.println("  ："+Thread.currentThread().getName()+"    body  ："+queue.take().getBody());
		    		 } catch (InterruptedException e) {
		      			e.printStackTrace();
		    	 	}
		   	}
	  	 }
	  }
 }

//  
 class  Message{
	  public Message(){}
	  private String topoic;
	  private String body;
	 
	  public String getTopoic() {
	   	return topoic;
	  }
	  public synchronized void setTopoic(String topoic) {
	   	this.topoic = topoic;
	  }
	  public String getBody() {
	  	 return body;
	  }
	  public synchronized void setBody(String body) {
	  	 this.body = body;
	  }
	  
}

}