4

再会。

我的网络爬虫项目有拦截器问题。逻辑很简单。首先创建一个Runnable，它下载 html 文档，扫描所有链接，然后在所有资助的链接上创建新Runnable对象。Runnable依次创建的每个新对象Runnable都会为每个链接创建新对象并执行它们。

问题是ExecutorService永远不会停止。

爬虫测试.java

public class CrawlerTest {

    public static void main(String[] args) throws InterruptedException {
        new CrawlerService().crawlInternetResource("https://jsoup.org/");
    }
}

爬虫服务.java

import java.io.IOException;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

public class CrawlerService {

    private Set<String> uniqueUrls = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>(10000));
    private ExecutorService executorService = Executors.newFixedThreadPool(8);
    private String baseDomainUrl;

    public void crawlInternetResource(String baseDomainUrl) throws InterruptedException {
        this.baseDomainUrl = baseDomainUrl;
        System.out.println("Start");
        executorService.execute(new Crawler(baseDomainUrl)); //Run first thread and scan main domain page. This thread produce new threads.
        executorService.awaitTermination(10, TimeUnit.MINUTES);
        System.out.println("End");
    }

    private class Crawler implements Runnable { // Inner class that encapsulates thread and scan for links

        private String urlToCrawl;

        public Crawler(String urlToCrawl) {
            this.urlToCrawl = urlToCrawl;
        }

        public void run() {
            try {
                findAllLinks();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        private void findAllLinks() throws InterruptedException {
            /*Try to add new url in collection, if url is unique adds it to collection, 
             * scan document and start new thread for finded links*/
            if (uniqueUrls.add(urlToCrawl)) { 
                System.out.println(urlToCrawl);

                Document htmlDocument = loadHtmlDocument(urlToCrawl);
                Elements findedLinks = htmlDocument.select("a[href]");

                for (Element link : findedLinks) {
                    String absLink = link.attr("abs:href");
                    if (absLink.contains(baseDomainUrl) && !absLink.contains("#")) { //Check that we are don't go out of domain
                        executorService.execute(new Crawler(absLink)); //Start new thread for each funded link
                    }
                }
            }
        }

        private Document loadHtmlDocument(String internetResourceUrl) {
            Document document = null;
            try {
                document = Jsoup.connect(internetResourceUrl).ignoreHttpErrors(true).ignoreContentType(true)
                        .userAgent("Mozilla/5.0 (Windows NT 6.1; WOW64; rv:48.0) Gecko/20100101 Firefox/48.0")
                        .timeout(10000).get();
            } catch (IOException e) {
                System.out.println("Page load error");
                e.printStackTrace();
            }
            return document;
        }
    }
}

这个应用程序需要大约 20 秒来扫描 jsoup.org 以获取所有唯一链接。但它只是等待 10 分钟executorService.awaitTermination(10, TimeUnit.MINUTES); ，然后我看到死的主线程并且仍在工作的执行程序。

线程

如何强制ExecutorService正常工作？

我认为问题在于它在另一个任务中而不是在主线程中调用 executorService.execute。

4

4 回答 4

3

你在滥用awaitTermination. 根据javadoc，您应该shutdown首先调用：

阻塞直到所有任务在关闭请求后完成执行，或者发生超时，或者当前线程被中断，以先发生者为准。

为了实现您的目标，我建议使用（或支持像这样CountDownLatch的增量的闩锁）来确定没有任务剩余的确切时刻，以便您可以安全地执行。shutdown

于 2016-08-12T10:27:05.990 回答
2

我看到你之前的评论：

我不能使用 CountDownLatch，因为我事先不知道我将从资源中收集多少个唯一链接。

首先，vsminkov 回答了为什么awaitTermniation要坐等 10 分钟。我将提供一个替代解决方案。

而不是CountDownLatch使用Phaser。对于每个新任务，您都可以注册并等待完成。

创建一个移相器，register每次execute.submit调用 a 并且arrive每次 aRunnable完成。

public void crawlInternetResource(String baseDomainUrl) {
    this.baseDomainUrl = baseDomainUrl;

    Phaser phaser = new Phaser();
    executorService.execute(new Crawler(phaser, baseDomainUrl)); 
    int phase = phaser.getPhase();
    phase.awaitAdvance(phase);
}

private class Crawler implements Runnable { 

    private final Phaser phaser;
    private String urlToCrawl;

    public Crawler(Phaser phaser, String urlToCrawl) {
        this.urlToCrawl = urlToCrawl;
        this.phaser = phaser;
        phaser.register(); // register new task
    }

    public void run(){
       ...
       phaser.arrive(); //may want to surround this in try/finally
    }
于 2016-08-12T18:52:59.407 回答
0

你不是叫关机。

这可能有效 - CrawlerService 中的 AtomicLong 变量。在每个新的子任务提交给执行器服务之前递增。

修改您的 run() 方法以减少此计数器，如果为 0，则关闭执行程序服务

public void run() {
    try {
        findAllLinks();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        //decrements counter
        //If 0, shutdown executor from here or just notify CrawlerService who would be doing wait().
    }
}

在“finally”中，减少计数器，当计数器为零时，关闭 executor 或只通知 CrawlerService。0 表示，这是最后一个，没有其他正在运行，队列中没有待处理。没有任务将提交任何新的子任务。

于 2016-08-12T12:42:00.387 回答
0

如何强制 ExecutorService 正常工作？

我认为问题在于它在另一个任务中而不是在主线程中调用 executorService.execute。

不，问题不在于 ExecutorService。您以不正确的方式使用 API，因此没有得到正确的结果。

您必须按一定顺序使用三个 API 才能获得正确的结果。

1. shutdown
2. awaitTermination
3. shutdownNow

来自ExecutorService的 oracle 文档页面的推荐方式：

 void shutdownAndAwaitTermination(ExecutorService pool) {
   pool.shutdown(); // Disable new tasks from being submitted
   try {
     // Wait a while for existing tasks to terminate
     if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
       pool.shutdownNow(); // Cancel currently executing tasks
       // Wait a while for tasks to respond to being cancelled
       if (!pool.awaitTermination(60, TimeUnit.SECONDS))
           System.err.println("Pool did not terminate");
     }
   } catch (InterruptedException ie) {
     // (Re-)Cancel if current thread also interrupted
     pool.shutdownNow();
     // Preserve interrupt status
     Thread.currentThread().interrupt();
   }

shutdown(): 启动有序关闭，其中执行先前提交的任务，但不会接受新任务。

shutdownNow():尝试停止所有正在执行的任务，停止等待任务的处理，并返回等待执行的任务列表。

awaitTermination():阻塞直到所有任务在关闭请求后完成执行，或者发生超时，或者当前线程被中断，以先发生者为准。

另一方面：如果您想等待所有任务完成，请参阅这个相关的 SE 问题：

等到所有线程在java中完成它们的工作

我更喜欢使用invokeAll()or ForkJoinPool()，它最适合您的用例。

于 2016-08-13T11:20:50.493 回答