多线程Java多线程执行超过循环边界

Question

我正在创建一个网络刮板，从网络上获取链接和电子邮件。这些链接将被用来寻找新的地方来搜索电子邮件，然后这些电子邮件将被存储在一个集合中。每个链接都被传递到自己线程中的固定线程池，以查找更多电子邮件。我一开始只查找10封电子邮件，但出于某种原因，我的代码返回了大约13封电子邮件

    while (emailSet.size() <= EMAIL_MAX_COUNT) {

        link = linksToVisit.poll();

        linksToVisit.remove(link);
        linksVisited.add(link);
        pool.execute(new Scraper(link));
    }

    pool.shutdownNow();

    emailSet.stream().forEach((s) -> {
        System.out.println(s);
    });
    System.out.println(emailSet.size());

虽然我知道有可能创建额外的线程，但在我收到10封电子邮件后，这些线程仍将运行，但不应该pool.shutdownNow()结束这些线程吗

这是我的线程代码，如果有帮助的话

class Scraper implements Runnable {

    private String link;

    Scraper(String s) {
        link = s;
    }

    @Override
    public void run() {
        try {
            Document doc = (Document) Jsoup.connect(link).get();

            Elements links = doc.select("a[href]");
            for (Element href : links) {
                String newLink = href.attr("abs:href");
                if (!linksVisited.contains(newLink) && !linksToVisit.contains(newLink)) {
                    linksToVisit.add(newLink);
                }
            }
            Pattern p = Pattern.compile(
                    "[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\\.[a-zA-Z0-9-.]+");
            Matcher matcher = p.matcher(doc.text());
            while (matcher.find()) {
                emailSet.add(matcher.group());
            }
        } catch (Exception e) {
            //Catch on of the many exceptions Jsoup.connect might throw 
            // and just let the thread expire.
        }
    }
}

编辑1：

我第一次应该包括这个，但我使用的是线程安全的集合和队列

Set<String> emailSet = Collections.synchronizedSet(new HashSet());
BlockingQueue<String> linksToVisit = new ArrayBlockingQueue(10000);
Set<String> linksVisited = Collections.synchronizedSet(new HashSet());
final int EMAIL_MAX_COUNT = 10;
ExecutorService pool = newFixedThreadPool(25);

编辑2

我想我应该用答案更新我的问题，所以这里是我的问题所在

    while (emailSet.size() <= EMAIL_MAX_COUNT) {

    link = linksToVisit.poll();

    linksToVisit.remove(link);
    linksVisited.add(link);
    pool.execute(new Scraper(link));
}

我的列表一开始只有一个链接。在第一个链接被删除后，我有一个空列表，它一直在创建没有链接可供搜索的新线程。在填充列表之前，我已经创建了数百个线程，它们什么都不做，只是减慢了系统的运行速度，直到它最终崩溃

下面是代码修复，以确保在没有搜索链接的情况下不会创建任何线程

while (emailSet.size() <= EMAIL_MAX_COUNT) {

        if (linksToVisit.size() > 0) {
            link = linksToVisit.poll();

            linksToVisit.remove(link);
            linksVisited.add(link);
            pool.execute(new Scraper(link));
            //System.out.println("Emails " + emailSet.size());
        } else {
            try {
                Thread.sleep(100);
            } catch (InterruptedException ex) {
                Logger.getLogger(Crawler.class.getName())
                        .log(Level.SEVERE, null, ex);
            }
        }
    }

Answer 1

在检查电子邮件集大小的循环中，您以异步方式启动scaper，但在循环的一个周期内，scraper可以找到多个电子邮件，或者您可以启动多个scraper，启动后，它会在页面上添加所有电子邮件链接 考虑下面的时间

T1 loop start ->T2 loop schedule Scaper ->T3 loop check emailSet ->T4 Scraper finds 13 email -> T5  loop check emailSet

还是下一个

T1 loop start ->T2 loop schedule Scaper "1" ->T3 loop check emailSet ->T4 loop schedule Scaper "2" T5 -> Scraper "1" finds 6 emails -> T6  loop check emailSet -> Scraper "1" finds 7 emails

等等

如果你想在发现10封电子邮件时停止，你必须更改以下一封

while (matcher.find()) {
    emailSet.add(matcher.group());
}

到

while (matcher.find()) {
    if (emailSet.size() <= EMAIL_MAX_COUNT) {
        emailSet.add(matcher.group());
    }
}

即使这样，也不能完全保证你可以在EMAIL_MAX_COUNT停止，因为多个线程（例如3个）可以检查大小，得到9个，然后所有线程都插入一封电子邮件

如果要确保电子邮件集的确切大小，必须在单个块内（使用synchronized(emailSet)或使用锁）同步读写操作；差不多

while (matcher.find()) {
    synchronized(emailSet) {
       if (emailSet.size() <= EMAIL_MAX_COUNT) {
           emailSet.add(matcher.group());
       }
    }
}