用Java解决生产者-消费者问题

当我们尝试多线程编程时，生产者-消费者问题是最常见的问题之一。尽管不像多线程编程中的其他一些问题那样具有挑战性，但是错误地实现此问题可能会造成应用程序混乱。生产的物品将不使用，开始的物品将被跳过，消耗量取决于生产是在消耗尝试之前还是之后开始的，等等。此外，您可能会在异常发生后很长时间注意到异常，最重要的是，几乎所有异常线程程序，这一程序也很难调试和复制。

因此，在这篇文章中，我认为我将尝试借助Java出色的java.util.concurrent包及其类来解决Java中的此问题。

首先，让我们看一下生产者－消费者问题的特征：

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

请注意，在此生产者－消费者问题中，生产者运行在与消费者不同的线程上。此设置在两种情况下有意义：

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

第二点中的“较大”一词有些宽松。考虑以下情况：生产者从文件中读取一行，而“消耗和处理”只是将行以特殊格式记录回文件中，那么使用生产者消费者问题解决方案可以被认为是过度设计的情况一个解法。但是，如果对于每行，“消耗和处理”步骤是向Web服务器发出HTTP GET / POST请求，然后将结果转储到某个地方，则我们应该选择生产者-消费者解决方案。在这种情况下，我假设行（item）本身具有执行GET / POST的所有数据，而我们不依赖于上一行/下一行。

因此，让我们首先看一下我在下面发布的生产者－消费者问题解决方案的特征：

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

有趣的是，要在通用级别解决此问题，我们只能解决消费者方，而不能解决生产方。这是因为项目的生产可以随时进行，而我们以通用方式进行项目生产的控制几乎没有。但是，我们可以在接受生产者提供的商品时控制消费者的行为。制定了规则之后，让我们看一下消费者合同：

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;}
}

上面唯一的挑战是while循环中的条件。这样编写while循环，即使在生产者完成生产并通知消费者生产完成之后，也可以支持项目消耗的继续。上面的while循环可确保在线程退出之前完成所有项目的消耗。

上面的使用者是线程安全的，可以共享多个生产者，以便每个生产者可以并发调用consumer.consume（），而不必担心同步和其他多线程警告。生产者只需要提交Item接口的实现，其process（）方法将包含如何完成消耗的逻辑。

作为阅读本文的奖励，我提出了一个测试程序，演示了如何使用上述类：

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中的生产者-消费者问题。