生产者消费者Java多线程实现

生产者和消费者问题是最经典的模拟多线程合作的问题之一，本文将提供一个具有多个生产者和多个消费者的Java实现。主要涉及到资源的锁定，多线程的合作，以及任务的抽象。

产品的抽象

每个产品都有一个唯一编号no，在构造函数里面初始化。

static class Product {
		private static int index = 0;
		private int no;

		public Product() {
			index++;
			no = index;
		}
	}

工厂的抽象

此处我用工厂来存储产品，可能用SuperMarket定义会更合适。多个生产者和多个消费者都共享同一个工厂，所以工厂必须保证线程安全，需要锁定。此处选择使用

java.util.concurrent.locks.Lock

来加锁，还可以选择synchronized关键字或者BlockedQueue来实现。工厂的存储用LinkedList，删除频繁，效率比ArrayList高。当有产品的时候，还需要通知(signalAll)等待购买的消费者。此外，工厂还能够输出自己的仓库统计信息。

static class Factory {
		private LinkedList<Product> productList = new LinkedList<Product>();
		private Lock lock = new ReentrantLock();
		private Condition condition = lock.newCondition();

		public void addProduct(Product product) {
			lock.lock();
			try {
				productList.offer(product);
				condition.signalAll();
			} finally {
				lock.unlock();
			}
		}

		public Product retrieveProduct() throws InterruptedException {
			lock.lock();

			try {
				while (productList.isEmpty())
					condition.await();
				return productList.poll();
			} finally {
				lock.unlock();
			}
		}

		@Override
		public String toString() {
			lock.lock();

			try {
				return "[Factory Info] " + productList.size() + " is stored.";
			} finally {
				lock.unlock();
			}
		}
	}

生产者的抽象

每一个生产者也有一个唯一编号no，且与一个factory关联。值得说明的是

Thread.interrupted()

能够响应

Thread.currentThread().interrupt()

。如果把sleep的时间改为随机数，应该更有意思。

static class Producer implements Runnable {
		private static int index = 0;
		private int no;
		private Factory factory;

		public Producer(Factory factory) {
			index++;
			no = index;
			this.factory = factory;
		}

		@Override
		public void run() {
			try {
				while (!Thread.interrupted()) {
					Product product = new Product();
					factory.addProduct(product);
					print("Producer " + no + " create product " + product.no);
					TimeUnit.SECONDS.sleep(1);
				}
			} catch (InterruptedException ex) {
				print("Producer " + no + " is interrupted.");
			}
		}
	}

消费者的抽象

和生产者一样，每一个消费者也有一个唯一编号no，且与一个factory关联。如果factory库存中没有产品，消费者会忠诚的等待在那里。

static class Consumer implements Runnable {
		private static int index = 0;
		private int no;
		private Factory factory;

		public Consumer(Factory factory) {
			index++;
			no = index;
			this.factory = factory;
		}

		@Override
		public void run() {
			try {
				while (!Thread.interrupted()) {
					Product product = factory.retrieveProduct();
					print("Consumer " + no + " consume product " + product.no);
					TimeUnit.SECONDS.sleep(1);
				}
			} catch (InterruptedException ex) {
				print("Consumer " + no + " is interrupted.");
			}
		}
	}

模拟测试

用Executors以及ExecutorService来执行生产和消费任务，简化了代码逻辑。有意思的是调整生产者数量以及消费者数量，查看factory的库存，库存会不同，可见生产与市场之间的供需关系，一定要弄好，否则会很麻烦。

public static void main(String[] args) {
		ExecutorService exe = Executors.newCachedThreadPool();

		final Factory factory = new Factory();

		for (int i = 0; i < 10; i++) {
			exe.execute(new Producer(factory));
		}

		for (int i = 0; i < 6; i++) {
			exe.execute(new Consumer(factory));
		}

		new Timer().schedule(new TimerTask() {
			@Override
			public void run() {
				print(factory);
			}
		}, 3000, 5000);
	}