java wait notifyAll 生产者 消费者 BlockingDeque

--用wait notifyAll来实现生产者与消费者模式，如下

package com.collonn.procon2;

import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicInteger;

public class PCTest {
	// the max number of product in product pool
	public static final int QUEUE_MAX_SIZE = 3;
	// product pool
	public static final LinkedList<Integer> QUEUE = new LinkedList<Integer>();
	// product name
	public static final AtomicInteger ATOMIC_INTEGER = new AtomicInteger();

	// the speed of producer
	public static final int PRODUCE_SPEED = 1000 * 1;
	// the speed of consumer
	public static final int CONSUME_SPEED = 100 * 1;

	// the number of producer
	public static final int PRODUCE_COUNT = 1;
	// the number of consumer
	public static final int CONSUME_COUNT = 5;

	// create producer
	private void produce() {
		for (int i = 0; i < PRODUCE_COUNT; i++) {
			new Thread(new Producer()).start();
		}
	}

	// create consumer
	private void consume() {
		for (int i = 0; i < CONSUME_COUNT; i++) {
			Thread thread = new Thread(new Consumer());
			thread.setName("td" + i);
			thread.start();
		}
	}

	public static void main(String[] args) throws Exception {
		PCTest pct1 = new PCTest();
		System.out.println("start producer...");
		pct1.produce();
		Thread.sleep(1000 * 5);
		System.out.println("start consumer...");
		pct1.consume();
	}

}

// producer
class Producer implements Runnable {

	@Override
	public void run() {
		while (true) {
			try {
				synchronized (PCTest.QUEUE) {
					if(PCTest.QUEUE.size() == PCTest.QUEUE_MAX_SIZE){
						System.out.println("product pool full...");
						PCTest.QUEUE.wait();
					}else{
						int next = PCTest.ATOMIC_INTEGER.getAndAdd(1);
						PCTest.QUEUE.addLast(next);
						System.out.println("produce," + next);

						PCTest.QUEUE.notifyAll();
					}
				}

				Thread.sleep(PCTest.PRODUCE_SPEED);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}

// consumer
class Consumer implements Runnable {

	@Override
	public void run() {
		try {
			while (true) {
				synchronized (PCTest.QUEUE) {
					if(PCTest.QUEUE.size() == 0){
						System.out.println("product pool empty...");
						PCTest.QUEUE.wait();
					}else{
						Integer data = PCTest.QUEUE.removeFirst();
						System.out.println("consume," + Thread.currentThread().getName() + "," + data);

						PCTest.QUEUE.notifyAll();
					}
				}

				Thread.sleep(PCTest.CONSUME_SPEED);
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

--用BlockingDeque来实现生产者与消费者模式，如下

package com.collonn.procon;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;

public class ProConMain {
	// product pool
	public static final BlockingDeque<Integer> BLOCK_QUEUE = new LinkedBlockingDeque<Integer>(5);
	// product name
	public static final AtomicInteger ATOMIC_INTEGER = new AtomicInteger();

	// the speed of producer
	public static final int PRODUCE_SPEED = 1000 * 3;
	// the speed of consumer
	public static final int CONSUME_SPEED = 1000 * 1;

	// the number of producer
	public static final int PRODUCE_COUNT = 1;
	// the number of consumer
	public static final int CONSUME_COUNT = 20;

	// create producer
	private void produce() {
		for (int i = 0; i < PRODUCE_COUNT; i++) {
			new Thread(new Producer()).start();
		}
	}

	// create consumer
	private void consume() {
		for (int i = 0; i < CONSUME_COUNT; i++) {
			Thread thread = new Thread(new Consumer());
			thread.setName("td" + i);
			thread.start();
		}
	}

	public static void main(String[] args) {
		ProConMain t1 = new ProConMain();
		t1.produce();
		t1.consume();
	}

}

// producer
class Producer implements Runnable {

	@Override
	public void run() {
		while (true) {
			try {
				int next = ProConMain.ATOMIC_INTEGER.getAndAdd(1);
				ProConMain.BLOCK_QUEUE.putLast(next);
				System.out.println("produce," + next);

				Thread.sleep(ProConMain.PRODUCE_SPEED);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}

// consumer
class Consumer implements Runnable {

	@Override
	public void run() {
		try {
			while (true) {
				Integer data = ProConMain.BLOCK_QUEUE.takeFirst();
				System.out.println("consume," + Thread.currentThread().getName() + "," + data);

				Thread.sleep(ProConMain.CONSUME_SPEED);
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

一：用Executor来实现生产者与消费者模式，如下

package com.collonn.executor;

import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ProConExecutor {
	// product name
	public static final AtomicInteger ATOMIC_INTEGER = new AtomicInteger();

	// the speed of producer
	public static final int PRODUCE_SPEED = 1000 * 1;
	// the speed of consumer
	public static final int CONSUME_SPEED = 1000 * 5;

	// the speed of consumer
	public static final int BLOCK_QUEUE_SIZE = 5;

	// create thread pool with deal strategy
	// corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, workQueue, rejectedExecutionHandler
	public static final Executor EXECUTOR = new ThreadPoolExecutor(1, 3, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(BLOCK_QUEUE_SIZE),
			new RejectedExecutionHandler() {
				@Override
				public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
					System.out.println("[overload], ignore data:" + ((Taskk)r).getData() + ", poolSize:" + e.getPoolSize() + ", queueSize:" + e.getQueue().size());
				}
			});

	public static void main(String[] args) {
		// produce product, then, throw to thead pool, then deal the data
		new Thread() {
			@Override
			public void run() {
				try {
					while (true) {
						int next = ProConExecutor.ATOMIC_INTEGER.getAndAdd(1);
						ProConExecutor.EXECUTOR.execute(new Taskk(next));
						System.out.println("produce," + next);

						Thread.sleep(PRODUCE_SPEED);
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}.start();
	}

}

// process data
class Taskk implements Runnable {
	private Integer data;

	public Taskk(Integer data) {
		this.data = data;
	}

	public Integer getData() {
		return data;
	}

	public void setData(Integer data) {
		this.data = data;
	}

	@Override
	public void run() {
		try {
			System.out.println("consume," + this.data);

			Thread.sleep(ProConExecutor.CONSUME_SPEED);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

总结：我们可以看到，代码在一步步的精简，且更优雅。

java wait notifyAll 生产者 消费者 BlockingDeque