• 生产者生产物品。
• 消费者消费生产者生产的物品。
• 生产者完成生产，并让消费者知道他们已经完成了。

• 消耗该项目的步骤独立产生，而不依赖于其他项目。
• 处理项目的时间大于生产项目的时间。

• 可以有多个生产者。
• 将有多个消费者。
• 一旦完成新物品的生产，生产者将告知消费者，以便消费者在消费并加工完最后一件物品后退出。

package com.maximus.producerconsumer;public interface Consumer
{public boolean consume(Item j);public void finishConsumption();
}

在这里，可以由多个类似商品的生产者共享消费者。 类似的项目，我的意思是生产者，其生产“项目”类型的对象。 Item的定义如下：

package com.maximus.consumer;public interface Item
{public void process();
}

现在我们来看一下Consumer接口的实现：

package com.maximus.consumer;import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;public class ConsumerImpl implements Consumer
{private BlockingQueue< Item > itemQueue = new LinkedBlockingQueue<Item>();private ExecutorService executorService = Executors.newCachedThreadPool();private List<ItemProcessor> jobList = new LinkedList<ItemProcessor>();private volatile boolean shutdownCalled = false;public ConsumerImpl(int poolSize){for(int i = 0; i < poolSize; i++){ItemProcessor jobThread = new ItemProcessor(itemQueue);jobList.add(jobThread);executorService.submit(jobThread);}}public boolean consume(Item j){if(!shutdownCalled){try{itemQueue.put(j);}catch(InterruptedException ie){Thread.currentThread().interrupt();return false;}return true;}else{return false;}}public void finishConsumption(){for(ItemProcessor j : jobList){j.cancelExecution();}executorService.shutdown();}
}

现在，唯一感兴趣的点是消费者内部用于处理传入商品的ItemProcessor。 ItemProcessor的编码如下：

package com.maximus.consumer;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;public class ItemProcessor implements Runnable
{private BlockingQueue<Item> jobQueue;private volatile boolean keepProcessing;public ItemProcessor(BlockingQueue<Item> queue){jobQueue = queue;keepProcessing = true;}public void run(){while(keepProcessing || !jobQueue.isEmpty()){try{Item j = jobQueue.poll(10, TimeUnit.SECONDS);if(j != null){j.process();}}catch(InterruptedException ie){Thread.currentThread().interrupt();return;}}}public void cancelExecution(){this.keepProcessing = false;}
}

package com.maximus.consumer;import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;public class Test
{public static void main(String[] args) throws Exception{Consumer consumer = new ConsumerImpl(10);BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(new File(args[0]))));String line = "";while((line = br.readLine()) != null){System.out.println("Producer producing: " + line);consumer.consume(new PrintJob(line));}consumer.finishConsumption();}
}class PrintJob implements Item
{private String line;public PrintJob(String s){line = s;}public void process(){System.out.println(Thread.currentThread().getName() + " consuming :" + line);}
}

参考： The Java HotSpot博客上的JCG合作伙伴 Sarma Swaranga 解决了Java中的生产者-消费者问题 。