使用ExecutorService并行处理作业

Question

我正在编写一个需要处理大量URL的java程序
每个URL将按顺序运行以下作业：下载、分析、压缩

我希望每个作业都有固定数量的线程，这样所有作业在任何给定时间都有并发运行的线程，而不是每个URL一次使用一个线程来完成所有作业。

例如，下载作业将有多个线程来获取和下载URL，一旦其中一个URL被下载，它就会将其传递给分析作业中的一个线程，一旦完成，它就会传递给压缩作业中的一个线程，等等。

我正在考虑在java中使用CompletionService，因为它一完成就返回一个结果，但我不确定它是如何工作的，到目前为止，我的代码如下所示：

ExecutorService executor = Executors.newFixedThreadPool(3);
CompletionService<DownloadedItem> completionService = new ExecutorCompletionService<DownloadedItem>(executor);

//while list has URL do {
   executor.submit(new DownloadJob(list.getNextURL());//submit to queue for download
//}

//while there is URL left do {
   Future<DownloadedItem> downloadedItem = executor.take();//take the result as soon as it finish
   //what to do here??
//}

我的问题是如何将下载的项目移动到分析作业并在那里完成工作而不等待所有下载作业完成？我正在考虑为每个作业创建一个CompletionService，这是一种可行的方法吗？如果不是，是否有更好的替代方法来解决这个问题？请提供示例。

Answer 1

您所描述的称为Pipeline。基本上下载任务的输出是分析任务的输入。分析的输出是压缩的输入。似乎有两个选项可以完成此操作：

1） 让下载任务知道输出的管道，这样它就可以自己提交结果。

class DownloadTask implement Runnable {
    Executor analyzePipeline;
    public void run() {
        //Do download stuff
        analyzePipeline.submit(new AnalyzeTask(downloaded content));
    }
}

2） 允许另一个线程将下载任务的结果移动到分析任务的管道中。

ExecutorService executor = Executors.newFixedThreadPool(3);
ExecutorService analyzeExecutor = Executors.newFixedThreadPool(3);
CompletionService<DownloadedItem> completionService = new ExecutorCompletionService<DownloadedItem>(executor);

while list has URL do {
   executor.submit(new DownloadJob(list.getNextURL());//submit to queue for download
}

new Thread() {
    public void run() {
        while there is URL left do {
            Future<DownloadedItem> downloadedItem = executor.take();//take the result as soon as it finish
            analyzeExecutor.submit(new AnalyzeJob(downloadedItem.get());
        }
    }
};    
//...and so on

Answer 2

你很接近。首先将任务提交给完成服务：

completionService.submit(new DownloadJob(list.getNextURL());

现在抓住未来并等待它：

DownloadedItem> downloadedItem = executor.take().get();

调用get（）可能会阻塞。重复上面的行，重复您提交的项目的次数。

如果您需要更大的吞吐量（在您的情况下，一次最多下载三个URL），请考虑async http client，它将允许您同时从数千个URL下载。它使用NIO，是事件驱动的，不涉及线程。

Answer 3

一旦您提到IN ORDER，任何为有序任务使用单独线程的尝试只会使系统的设计复杂化。

在我看来，最好的办法是让不同的线程同时处理不同的URL。要完成这三个步骤，可以引入另一个抽象（比如使用3个可调用项），但仍然希望在一个线程中顺序执行它们。无需完成服务。