Java生产者消费者模式实战解析

异步模式

传统版

异步模式之生产者/消费者：

classShareData{privateintnumber=0;privateLocklock=newReentrantLock();privateConditioncondition=lock.newCondition();publicvoidincrement()throwsException{// 同步代码块，加锁lock.lock();try{// 判断 防止虚假唤醒while(number!=0){// 等待不能生产condition.await();}// 干活number++;System.out.println(Thread.currentThread().getName()+"\t "+number);// 通知 唤醒condition.signalAll();}catch(Exceptione){e.printStackTrace();}finally{lock.unlock();}}publicvoiddecrement()throwsException{// 同步代码块，加锁lock.lock();try{// 判断 防止虚假唤醒while(number==0){// 等待不能消费condition.await();}// 干活number--;System.out.println(Thread.currentThread().getName()+"\t "+number);// 通知 唤醒condition.signalAll();}catch(Exceptione){e.printStackTrace();}finally{lock.unlock();}}}publicclassTraditionalProducerConsumer{publicstaticvoidmain(String[]args){ShareDatashareData=newShareData();// t1线程，生产newThread(()->{for(inti=0;i<5;i++){shareData.increment();}},"t1").start();// t2线程，消费newThread(()->{for(inti=0;i<5;i++){shareData.decrement();}},"t2").start();}}

改进版

异步模式之生产者/消费者：

• 消费队列可以用来平衡生产和消费的线程资源，不需要产生结果和消费结果的线程一一对应
• 生产者仅负责产生结果数据，不关心数据该如何处理，而消费者专心处理结果数据
• 消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据
• JDK 中各种阻塞队列，采用的就是这种模式

publicclassdemo{publicstaticvoidmain(String[]args){MessageQueuequeue=newMessageQueue(2);for(inti=0;i<3;i++){intid=i;newThread(()->{queue.put(newMessage(id,"值"+id));},"生产者"+i).start();}newThread(()->{while(true){try{Thread.sleep(1000);Messagemessage=queue.take();}catch(InterruptedExceptione){e.printStackTrace();}}},"消费者").start();}}//消息队列类，Java间线程之间通信classMessageQueue{privateLinkedList<Message>list=newLinkedList<>();//消息的队列集合privateintcapacity;//队列容量publicMessageQueue(intcapacity){this.capacity=capacity;}//获取消息publicMessagetake(){//检查队列是否为空synchronized(list){while(list.isEmpty()){try{sout(Thread.currentThread().getName()+":队列为空，消费者线程等待");list.wait();}catch(InterruptedExceptione){e.printStackTrace();}}//从队列的头部获取消息返回Messagemessage=list.removeFirst();sout(Thread.currentThread().getName()+"：已消费消息--"+message);list.notifyAll();returnmessage;}}//存入消息publicvoidput(Messagemessage){synchronized(list){//检查队列是否满while(list.size()==capacity){try{sout(Thread.currentThread().getName()+":队列为已满，生产者线程等待");list.wait();}catch(InterruptedExceptione){e.printStackTrace();}}//将消息加入队列尾部list.addLast(message);sout(Thread.currentThread().getName()+":已生产消息--"+message);list.notifyAll();}}}finalclassMessage{privateintid;privateObjectvalue;//get set}

阻塞队列

publicstaticvoidmain(String[]args){ExecutorServiceconsumer=Executors.newFixedThreadPool(1);ExecutorServiceproducer=Executors.newFixedThreadPool(1);BlockingQueue<Integer>queue=newSynchronousQueue<>();producer.submit(()->{try{System.out.println("生产...");Thread.sleep(1000);queue.put(10);}catch(InterruptedExceptione){e.printStackTrace();}});consumer.submit(()->{try{System.out.println("等待消费...");Integerresult=queue.take();System.out.println("结果为:"+result);}catch(InterruptedExceptione){e.printStackTrace();}});}