차단 대기 열 (BlockingQueue) 로 읽 기와 쓰기 (생산자 소비자) 스 레 드 구현

package concurrent;


import java.util.Scanner;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;


/**
 * @author: kimt
 * @Version: 1.0
 * @date: 2017 11 12    8:48:32
 * @modify:
 * @Description:        BlockingQueue(    )
 */
public class ProducerAndComsumer {


public static void main(String[] args) {
BlockingQueue bQueue = new LinkedBlockingQueue<>();
//   cached   
ExecutorService service = Executors.newFixedThreadPool(2);
//           
WriteThread producer = new WriteThread(bQueue);
ReadThread consumer = new ReadThread(bQueue);
//     
//        ，        ，       
service.execute(consumer);
service.execute(producer);
}


}


/**
 * 
 *      ( )  
 *
 */
class WriteThread implements Runnable {
private BlockingQueue bq;
private boolean isRunning = false;


public WriteThread(BlockingQueue bq) {
this.bq = bq;
}


@Override
public void run() {
isRunning = true;
Scanner sr = null;
while (isRunning) {
System.out.println(Thread.currentThread()+"     :");
//        
sr = new Scanner(System.in);
StringBuffer sb = new StringBuffer();
try {
//            
if (sr.hasNextLine()) {
sb.append(sr.nextLine());
}
//           
bq.put(sb.toString());
} catch (InterruptedException e) {
shutdown();
e.printStackTrace();
} 
}
}


public void shutdown() {
isRunning = false;
}


}


/**
 * 
 *      ( )  
 *
 */
class ReadThread implements Runnable {
private BlockingQueue bq;
private boolean isRunning = false;


public ReadThread(BlockingQueue bq) {
this.bq = bq;
}


@Override
public void run() {
isRunning = true;
while (isRunning) {
try {
//           ，        
String result = bq.take();
System.out.println(Thread.currentThread()+"     :" + result);
} catch (InterruptedException e) {
shutdown();
e.printStackTrace();
}
}
}


public void shutdown() {
isRunning = false;
}


}