| package concurrency; import java.util.LinkedList; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class ProducerAndConsumer { /** * 使用wait notify 的多生產(chǎn)者多消費(fèi)者,固定消費(fèi)空間的例子 * 消費(fèi)者 生產(chǎn)者的 執(zhí)行速度不一 */ private List<Integer> data = new LinkedList<>(); private static final int MAX_DATA_LEN = 10; class Producer implements Runnable { private int pid = 0; public Producer(int pid){ this.pid = pid; } public void run() { try { while (!Thread.currentThread().isInterrupted()) { synchronized (data) { //try { while (data.size() >= MAX_DATA_LEN) { System.out.println('Producer' pid ' waiting ! size : ' data.size()); data.wait(); } data.add(pid); System.out.println('Producer' pid ' add ' pid ' size: ' data.size()); data.notifyAll(); } Thread.sleep(50); } } catch (InterruptedException ie) { ie.printStackTrace(); } } } class Consumer implements Runnable{ private int cid = 0; public Consumer(int cid){ this.cid = cid; } public void run(){ try { while (!Thread.currentThread().isInterrupted()) { synchronized (data) { while (data.isEmpty()) { System.out.println('Consumer' cid ' waiting, data size : ' data.size()); data.wait(); } int pid = data.remove(0); System.out.println('Consumer' cid ' consuming data ' pid ' data size : ' data.size()); data.notifyAll(); } Thread.sleep(500); } }catch (InterruptedException ie){ ie.printStackTrace(); } } } public void start(){ ExecutorService executor = Executors.newCachedThreadPool(); for(int i = 0; i < 5; i){ executor.submit(new Producer(i)); executor.submit(new Consumer(i)); } try { Thread.sleep(10*1000); } catch (InterruptedException e) { e.printStackTrace(); } executor.shutdownNow(); } public static void main(String []args){ new ProducerAndConsumer().start(); } } |
|
|
