Java用BlockingQueue来实现简单的消息队列（生产者与消费者模式）

喜欢ヅ旅行 2023-07-14 10:51 12阅读 0赞

学习过程，当做笔记记录下来了：

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;
    	  }
    	  
    }
    
    }